pyspark 集成jupyter与pyspark on yarn-摩杜云开发者社区

标签（空格分隔）： Spark的部分

一：安装jupyter

Hadoop 集群 + spark 集群安装忽略


yum install epel-release 

yum install python36 


pip3 install --upgrade pip  # 升级pip 到最新版本

pip3 install jupyter  # 安装 jupyter 

jupyter notebook --generate-config   # 安装后先生成配置文件，用于后面写入ip，端口号，密码等

ipython # 进入python终端 
ipython
In [1]: from notebook.auth import passwd      #导入包
In [2]: passwd()                              #调用包
Enter password:                              #输入密码，用于后面登陆jupyter notebook
Verify password:                              #确认密码
Out[2]: 'sha1:ce23d945972f:34769685a7ccd3d08c84a18c63968a41f1140274'  #这段是密钥，复制下来等会用
exit() #退出


cd /root/

vim /root/.jupyter/jupyter_notebook_config.py

----
#在任意地方加上如下内容（键入i进入插入模式）
c.NotebookApp.ip = '*'  #允许任何ip去访问我们的jupyter notebook
c.NotebookApp.password = u'sha1:ce23d945972f:34769685a7ccd3d08c84a18c63968a41f1140274'  #u后面加上你复制的密钥
c.NotebookApp.open_browser = False    #因为是在linux，就不让他打开浏览器了
c.NotebookApp.port = 8888    #随便指定一个端口，如果这个端口被占用，jupyter会加1重新找端口，
                             直到找到为止
c.NotebookApp.allow_remote_access = True  #允许远程控制
c.NotebookApp.notebook_dir = u'路径'  #设置你打开jupyter notebook的时候想显示的位置/root/.jupyter，
                                      可以设置成经常使用的路径
#配置完记得保存（键入esc，然后输入:wq保存并退出）
----

启动jupyter notebook

# jupyter notebook

在windows下远程访问jupyter notebook，在本地打开浏览器访问以下地址：

http://172.16.30.10.31:8888

就可以打开jupyter  显示页面 


然后卸载python36 

yum remove -y python36 


安装anconda  [集群说有节点需要安装]

chmod +x Anaconda3-2022.10-Linux-x86_64.sh

./Anaconda3-2022.10-Linux-x86_64.sh

安装到 /opt/anaconda3/ 这个目录 

然后 写入环境变量

vim /etc/profile 
-----
export PATH=/opt/anaconda3/bin:$PATH
------

source /etc/profile 

conda -V 
------

# python3  执行命令保证能进入python3 的 终端

------
# 安装pyspark 插件

pip3 install -i https://pypi.tuna.tsinghua.edu.cn/simple pyspark  


vim /etc/profile
----
# Spark安装目录
#export SPARK_HOME=/opt/bigdata/spark
#export PATH=$PATH:$SPARK_HOME/sbin:$SPARK_HOME/bin:$PATH
#指向spark目录下的python文件夹和py4j包
export PYTHON_PATH=$SPARK_HOME/python:$SPARK_HOME/python/lib/py4j-0.10.9.5-src.zip:$PYTHON_PATH
#使用python3
export PYSPARK_PYTHON=/opt/anaconda3/bin/python3
export PYSPARK_DRIVER_PYTHON=/opt/anaconda3/bin/python3
#export PYSPARK_DRIVER_PYTHON=jupyter
#export PYSPARK_DRIVER_PYTHON_OPTS="notebook --ip=172.30.10.31 --no-browser --allow-root"
----

cd /opt/bigdata/spark/
bin/pyspark --master spark://172.30.10.31:7077   ### 这里将启动一个 spark任务

保证能够调起spark的集群master任务：

-----
>>> lines = sc.textFile("file:///usr/local/spark/sparksqldata/Data01.txt")
>>> res = lines.map(lambda x:x.split(",")).map(lambda x: x[0]) //获取每行数据的第1列 
>>> distinct_res = res.distinct()  //去重操作
>>> distinct_res.count()//取元素总个数
256 
-----
hdfs dfs -mkdir /input 
hdfs dfs -put Data01.txt /input 
>>> lines = sc.textFile("hdfs://hadoop01:8020/input/Data01.txt")
>>> res = lines.map(lambda x:x.split(",")).map(lambda x: x[0]) //获取每行数据的第1列 
>>> distinct_res = res.distinct()  //去重操作
>>> distinct_res.count()//取元素总个数
---
256
----


配置spark 与 jupyter 终端集成：

vim /etc/profile

---
# Spark安装目录
#export SPARK_HOME=/opt/bigdata/spark
#export PATH=$PATH:$SPARK_HOME/sbin:$SPARK_HOME/bin:$PATH
#指向spark目录下的python文件夹和py4j包
export PYTHON_PATH=$SPARK_HOME/python:$SPARK_HOME/python/lib/py4j-0.10.9.5-src.zip:$PYTHON_PATH
#使用python3
export PYSPARK_PYTHON=/opt/anaconda3/bin/python3
#export PYSPARK_DRIVER_PYTHON=/opt/anaconda3/bin/python3
export PYSPARK_DRIVER_PYTHON=jupyter
export PYSPARK_DRIVER_PYTHON_OPTS="notebook --ip=172.30.10.31 --no-browser --allow-root"
export PATH=/opt/anaconda3/bin:$PATH

----

nohup pyspark --master spark://hadoop01:7077 > jupyter-web.logs &

在这个地方有jupyter-web.log 有jupyer的打开的web 地址 
比如：
      http://172.30.10.31:8891/notebooks/

----
#from pyspark import SparkContext
#sc = SparkContext("local","count app")
words = sc.parallelize(
["scala",
"java",
"hadoop"
"spark",
"python",
"C++",
"pyspark"]
)
words_map = words.map(lambda x:(x,1))
mapping = words_map.collect()
print("key value pair -> %s" % (mapping))
----

----
lines = sc.textFile("hdfs://hadoop01:8020/input/Data01.txt")
res = lines.map(lambda x:x.split(",")).map(lambda x: x[0]) //获取每行数据的第1列 
distinct_res = res.distinct()  //去重操作
distinct_res.count()//取元素总个数
-----

二： pyspark on yarn 的设置

pyspark on yarn 实例：

vim file.csv 
---
id,value
1,10.0
2,20.0
3,30.0
4,40.0
5,50.0
---
hdfs dfs -mkdir /root 

hdfs dfs -put file.csv /root

hdfs dfs -chmod 777 -R /root

hdfs dfs -mkdidr /root/filecount/

-----

vim script.py 
----
from pyspark.sql import SparkSession

# 创建一个SparkSession对象
spark = SparkSession.builder.appName("MyApp").getOrCreate()

# 从HDFS中读取数据
df = spark.read.format("csv").option("header", "true").load("/root/file.csv")

# 执行一些数据处理操作
count = df.count()
avg = df.select("value").agg({"value": "avg"}).collect()[0][0]

# 将结果写入HDFS中
output = spark.createDataFrame([(count, avg)], ["count", "avg"])
output.write.format("csv").mode("overwrite").save("/root/filecount/")

# 关闭SparkSession对象
spark.stop()
-----

#使用python3 当前环境变量执行：
export PYSPARK_PYTHON=/opt/anaconda3/bin/python3
export PYSPARK_DRIVER_PYTHON=/opt/anaconda3/bin/python3

###spark-submit 提交：

spark-submit --master yarn --deploy-mode cluster --num-executors 10 --executor-cores 2 --executor-memory 2G /root/script.py

一直到 任务结束

hdfs dfs -get /root/filecount/

然后查看统计结果。