问题
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-filesargument ofspark-submitto add.py,.zipor.egg
files to be distributed with your application. If you depend on multiple Python files we recommend
packaging them into a.zipor.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
