PySpark on Yarn的相关依赖的解决方式

问题

Spark on Yarn是将yarn作为ClusterManager的运行模式，Spark会将资源（container）的管理与协调统一交给yarn去处理。

Spark on Yarn分为client/cluster模式：
对于client模式，Spark程序的Driver/SparkContext实例用户提交机上，该机器可以位于yarn集群之内或之外，只需要起能正常与ResourceManager通信及正确配置HADOOP_CONF_DIR或YARN_CONF_DIR环境变量指向yarn集群。生产环境中，通常提交机不会是yarn集群内部的节点，手握配置权限的情况下，可以按需配置支撑Spark程序需要的软件、环境、文件等。
对于cluster模式，Spark程序的Driver/SparkContext实例位于ApplicationMaster（am）中，am作为一个container可以起在yarn集群中任何一个NodeManager上，默认情况下，我们就需要为所有的节点机器准备好Spark程序需要的所有运行环境。

Python提供了非常丰富的数学运算、机器学习处理库——如numpy、pandas、scipy等等。越来越多的同事希望利用这些高效的库开发各种算法然后以PySpark程序跑到我们的Spark上。

对于scala/java写的Spark程序，我们可以将我们所依赖的jar一起与我们的main函数所在的主程序打成一个fat jar，通过spark-submit提交后，这些依赖就会通过Yarn的Distribute Cache分发到所有节点支撑运行。
对于python写的Spark程序如果有外部依赖就很尴尬了，python本身就是两种语言，在所有NodeManager节点上安装你所有需要的依赖对于IT运维人员也是一个非常痛苦的事情。

参考官方文档

For Python, you can use the --py-files argument of spark-submit to add .py, .zip or .egg
files to be distributed with your application. If you depend on multiple Python files we recommend
packaging them into a .zip or .egg.

--py-files，可以解决部分依赖的问题，但对于有些场景就可能不是很方便，或者不可能实现。

• 依赖太多，包括传递依赖
• python包在deploy前需要依赖的C代码提前编译
• 基于不同版本的python的pyspark跑在同一个yarn集群上

对于这些问题 ，社区也有相关的讨论，详细可以看下 这个ticket https://issues.apache.org/jira/browse/SPARK-13587

原理

pyspark原理的资料比较少，可以看下wiki

https://cwiki.apache.org/confluence/display/SPARK/PySpark+Internals

可以看下上面链接中的图，图中左右分为driver/executor, 图白色和绿色分python和java，可以看到不管PySpark适宜client还是cluster模式跑在yarn上，driver和executor端都有python的进程起着，这就需要集群中的所有节点都有相应的python依赖环境。

方案

从灵活性的角度来讲，这里从前辈的讨论中总结一下，提供一种在运行时创建python运行及相关依赖的办法

1、下载并安装anaconda
https://www.anaconda.com/download/#linux

2、安装anaconda

sh Anaconda2-5.0.1-Linux-x86_64.sh

2、创建需要的依赖环境conda create

conda create --name py2env_take1 --quiet  --copy --yes python=2 numpy scipy pandas

打印如下消息

Package plan for installation in environment /home/hadoop/anaconda2/envs/py2env_take1:

The following NEW packages will be INSTALLED:

    ca-certificates: 2017.08.26-h1d4fec5_0
    certifi:         2017.11.5-py27h71e7faf_0
    intel-openmp:    2018.0.0-hc7b2577_8
    libedit:         3.1-heed3624_0
    libffi:          3.2.1-hd88cf55_4
    libgcc-ng:       7.2.0-h7cc24e2_2
    libgfortran-ng:  7.2.0-h9f7466a_2
    libstdcxx-ng:    7.2.0-h7a57d05_2
    mkl:             2018.0.1-h19d6760_4
    ncurses:         6.0-h9df7e31_2
    numpy:           1.13.3-py27hbcc08e0_0
    openssl:         1.0.2m-h26d622b_1
    pandas:          0.21.0-py27he307072_1
    pip:             9.0.1-py27ha730c48_4
    python:          2.7.14-hdd48546_24
    python-dateutil: 2.6.1-py27h4ca5741_1
    pytz:            2017.3-py27h001bace_0
    readline:        7.0-ha6073c6_4
    scipy:           1.0.0-py27hf5f0f52_0
    setuptools:      36.5.0-py27h68b189e_0
    six:             1.11.0-py27h5f960f1_1
    sqlite:          3.20.1-hb898158_2
    tk:              8.6.7-hc745277_3
    wheel:           0.30.0-py27h2bc6bb2_1
    zlib:            1.2.11-ha838bed_2

第一次根据网络情况下载上述这些依赖，可能会比较久，以后就会快很多。

du -sh ~/anaconda2/envs/py2env_take1/
965M    /home/hadoop/anaconda2/envs/py2env_take1/

可以看到依赖包整个大小还是挺大的，对于一些实时性比较高的场景这种方式其实不太有利，有些不需要的依赖在创建的时候可以不打进去。当然我们还需要zip压缩一下，可以减小部分网络开销。当然如果我们把这个环境直接提前put到hdfs，也就没有这个问题了。

zip -r -9 -q py2env_take1.zip ./py2env_take1/

du -sh py2env_take1.zip
345M    py2env_take1.zip

这样我们就可以通过--archives path/to/py2env_take1.zip#py2env的方式将python环境上传并分发到spark各个进程的working dir。

测试

不会写python，从spark示例代码里拷一个出来玩玩

from __future__ import print_function

import numpy as np

from pyspark import SparkContext
# $example on$
from pyspark.mllib.stat import Statistics
# $example off$

if __name__ == "__main__":
    sc = SparkContext(appName="CorrelationsExample")  # SparkContext

    # $example on$
    seriesX = sc.parallelize([1.0, 2.0, 3.0, 3.0, 5.0])  # a series
    # seriesY must have the same number of partitions and cardinality as seriesX
    seriesY = sc.parallelize([11.0, 22.0, 33.0, 33.0, 555.0])

    # Compute the correlation using Pearson's method. Enter "spearman" for Spearman's method.
    # If a method is not specified, Pearson's method will be used by default.
    print("Correlation is: " + str(Statistics.corr(seriesX, seriesY, method="pearson")))

    data = sc.parallelize(
        [np.array([1.0, 10.0, 100.0]), np.array([2.0, 20.0, 200.0]), np.array([5.0, 33.0, 366.0])]
    )  # an RDD of Vectors

    # calculate the correlation matrix using Pearson's method. Use "spearman" for Spearman's method.
    # If a method is not specified, Pearson's method will be used by default.
    print(Statistics.corr(data, method="pearson"))
    # $example off$

    sc.stop()

take1

bin/spark-submit --master yarn --deploy-mode client --proxy-user hzyaoqin  /home/hadoop/data/apache-spark/spark-2.1.2-bin-hadoop2.7/examples/src/main/python/mllib/correlations_example.py
17/11/24 23:51:29 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Traceback (most recent call last):
  File "/home/hadoop/data/apache-spark/spark-2.1.2-bin-hadoop2.7/examples/src/main/python/mllib/correlations_example.py", line 20, in <module>
    import numpy as np
ImportError: No module named numpy

client模式，提示未安装numpy，失败

take2

bin/spark-submit --master yarn --deploy-mode cluster --proxy-user hzyaoqin  /home/hadoop/data/apache-spark/spark-2.1.2-bin-hadoop2.7/examples/src/main/python/mllib/correlations_example.py

cluster 模式，失败
查看am日志

Log Type: stdout
Log Upload Time: Fri Nov 24 23:49:11 +0800 2017
Log Length: 148
Traceback (most recent call last):
  File "correlations_example.py", line 20, in <module>
    import numpy as np
ImportError: No module named numpy

take3

bin/spark-submit --master yarn --deploy-mode cluster --proxy-user hzyaoqin --archives ~/anaconda2/envs/py2env_take2.zip#python2env  --conf spark.yarn.appMasterEnv.PYSPARK_PYTHON=python2env/py2env_take2/bin/python /home/hadoop/data/apache-spark/spark-2.1.2-bin-hadoop2.7/examples/src/main/python/mllib/correlations_example.py

输出结果，成功

Log Type: stdout
Log Upload Time: Fri Nov 24 23:47:45 +0800 2017
Log Length: 149
Correlation is: 0.850028676877
[[ 1.          0.97888347  0.99038957]
 [ 0.97888347  1.          0.99774832]
 [ 0.99038957  0.99774832  1.        ]]

测试版本

https://www.apache.org/dyn/closer.lua/spark/spark-2.2.0/spark-2.2.0-bin-hadoop2.7.tgz
https://www.apache.org/dyn/closer.lua/spark/spark-2.1.2/spark-2.1.2-bin-hadoop2.7.tgz
https://www.apache.org/dyn/closer.lua/spark/spark-2.0.2/spark-2.0.2-bin-hadoop2.7.tgz

其他版本未测试

结论

1、依靠anaconda 创建python依赖环境
2、通过--archives 上传该环境
3、通过spark.yarn.appMasterEnv.PYSPARK_PYTHON指定python执行目录
4、cluster模式可以，client模式显式指定PYSPARK_PYTHON，会导致PYSPARK_PYTHON环境变量不能被spark.yarn.appMasterEnv.PYSPARK_PYTHON overwrite
5、如果executor端也有numpy等依赖，应该要指定spark.executorEnv.PYSPARK_PYTHON(I guess)
6、改日试下anaconda3 创建python3的隔离环境试下。

参考

https://github.com/massmutual/sample-pyspark-application

其他

关于Client模式下的问题，提了个PR，欢迎讨论
https://github.com/apache/spark/pull/19840