pyspark 命令不执行_PySpark、Pyflink on jupyter

分手后的思念是犯贱 2022-10-26 12:47 215阅读 0赞

**JupyterNotebook** 是一个基于web的交互式开发环境。适合探索式开发以、PoC、数据分析数据挖掘等场景。目前支持运行40 多种编程语言，包括Python, R, Julia, and Scala。

pyspark，apache spark提供的python 开发接口。

pyflink， apache flink提供的python 开发接口。

## 1.相关性python基础知识 ##

**python环境变量**

*  PYTHONPATH，python的搜索路径，在python中import一个模块，是在这个环境变量定义的目录下查找的。
 *  PYTHONSTARTUP，Python启动后，先寻找PYTHONSTARTUP环境变量，然后执行此变量指定的python文件。

## 2. 安装pyspark和pyflink ##

最好先建立一个虚拟环境, pyspark和pyflink使用的py4j版本冲突，但是并不影响使用。

⚠️：pyflink的package名是“apache-flink”,。

# 创建虚拟环境
    python3 -m venv pybigdata
    
    cd pybigdata
    #启动虚拟环境
    source bin/activate
    #安装pyspark
    pip install pyspark
    #安装pyflink，注意这个包名
    pip install apache-flink
    #安装jupyter
    pip install jupyter

## 3.pyspark on jupyter ##

pyspark安装完成后，在虚拟环境的bin目录下会添加若干相关文件，我们最关心的的一个文件是“pyspark”。

这里有几个关键的环境变量：

*  SPARK\_HOME，如果设置了这个环境变量，
 *  PYSPARK\_DRIVER\_PYTHON，spark的driver程序使用的python命令
 *  PYSPARK\_PYTHON，spark节点调用的python命令
 *  SPARK\_HOME，如果设置了这个环境变量，HOME\}/python/”目录和“$\{SPARK\_HOME\}/python/lib/py4j-0.10.7-src.zip”
 *  PYSPARK\_DRIVER\_PYTHON\_OPTS，传递给PYSPARK\_DRIVER\_PYTHON的额外参数。
 *  PYTHONSTARTUP，设置成了“$\{SPARK\_HOME\}/python/pyspark/shell.py”，即在执行pyspark，进入python命令被执行时，要先调用这个脚本，初始化spark环境。

要在pyspark文件中添加环境变量：“PYSPARK\_DRIVER\_PYTHON=jupyter”和“PYSPARK\_DRIVER\_PYTHON\_OPTS=notebook”。主要添加位置，在调用完find-spark-home，设置才有效。

#要在这个代码片段下面添加环境变量
    if [ -z "${
         SPARK_HOME}" ]; then
      source "$(dirname "$0")"/find-spark-home
    fi
    #这是要添加的环境变量
    PYSPARK_DRIVER_PYTHON=jupyter
    PYSPARK_DRIVER_PYTHON_OPTS=notebook

环境变量配置完成后，执行“pyspark”，浏览器会打开一下界面：

![896f093bf78d98da3e85bdb3906bd188.png][]

新建一个notebook，notebook中绑定了如下变量：sc、spark、sql、sqlContext、sqlCtx。

1.sc变量是一个sparkcontext对象2

![99ec7ba0a8e1b71686471376c85648f8.png][]

2. spark变量是一个sparksession对象

![6ee6635dd9fd3a60d91c3256b49682ee.png][]

3. sql变量是sparksession的sql方法

![f8ed19a9ca1d9dc407c1270b904287ff.png][]

4. sqlCtx和sqlContext是同一个sqlcontext对象

![22f849db99d20cc483672c0777402aa6.png][]

至此就可以使用jupyter以及pyspark进行开发了。

4. pyflink on jupyter

安装完pyflink之后，在虚拟环境的bin目录下会添加若干相关文件，我们最关心的的一个文件是“pyflink-shell.sh”。最后启动代码如下：

# -i: 交互式
    # -m: 执行pyflink模块下的shell.py
    ${PYFLINK_PYTHON} -i -m pyflink.shell

将代码修改为：

# -i: 交互式
    # -m: 执行pyflink模块下的shell.py
    ${
         PYFLINK_PYTHON} -i -m pyflink.shell
    export PYTHONSTARTUP=$FLINK_HOME/shell.py
    jupyter notebook

修改$FLINK\_HOME/shell.py。将文件的第33行注释掉，下面添加一行“utf8\_out = sys.stdout”。代码如下：

# utf8_out = open(sys.stdout.fileno(), mode='w', encoding='utf8', buffering=1)
    utf8_out = sys.stdout

> $FLINK\_HOME/shell.py实际的目录是：“ *pybigdata/lib/python3.8/site-packages”*。pybigdata是虚拟环境文件夹。

执行如下命令，启动jupyter notebook

# unset这个步骤必须有
    # 删除已经存在的FLINK_HOME环境变量
    unset FLINK_HOME
    pyflink-shell.sh local

打开新的notebook,环境中绑定了四个变量

*  批处理表环境变量名为： bt\_env（pyflink.table.table\_environment.BatchTableEnvironment）
 *  比处理执行环境变量名为：b\_env（pyflink.dataset.execution\_environment.ExecutionEnvironment）
 *  流表环境变量名为：st\_env(pyflink.table.table\_environment.StreamTableEnvironment)
 *  流执行环境变量名为：s\_env(pyflink.datastream.stream\_execution\_environment.StreamExecutionEnvironment)

![8f5c15b8c99a28c60cd1e6700f058ba9.png][]

[896f093bf78d98da3e85bdb3906bd188.png]: /images/20221024/3be3d1fb45614c7193b4ae69fddb38c8.png
[99ec7ba0a8e1b71686471376c85648f8.png]: /images/20221024/39231ad71b4140d4a7b19c45ba495ed5.png
[6ee6635dd9fd3a60d91c3256b49682ee.png]: /images/20221024/99a89641d06b45e482e49a6de916f274.png
[f8ed19a9ca1d9dc407c1270b904287ff.png]: /images/20221024/e491dd1a6fe248fe96030d229c449111.png
[22f849db99d20cc483672c0777402aa6.png]: /images/20221024/4e3c01422de1475cb979e88593221a2e.png
[8f5c15b8c99a28c60cd1e6700f058ba9.png]: /images/20221024/c4e3c743f1554a77859fd3ebf6c6ac3a.png