为PySpark绑定Python3包会导致缺少导入

Question

我试图运行一个依赖于某些python3库的PySpark作业。我知道我可以在Spark Cluster上安装这些库，但是因为我正在为多个作业重用集群，所以我更愿意捆绑所有依赖项，并通过--py文件指令将它们传递给每个作业。

为此，我使用：

pip3 install -r requirements.txt --target ./build/dependencies
cd ./build/dependencies
zip -qrm . ../dependencies.zip

它有效地压缩了需要在根级别使用的包中的所有代码。

在mymain.py中，我可以导入依赖项

if os.path.exists('dependencies.zip'):
    sys.path.insert(0, 'dependencies.zip')

并将.zip添加到我的Spark上下文中

sc.addPyFile('dependencies.zip')

到目前为止还不错。

但出于某种原因，这将在星火星团的某种依赖地狱中恶化

跑步

spark提交--py files dependencies.zip main.py

在main.py（或类）中，我想使用熊猫。它将触发此错误的代码：

回溯（最近一次呼叫最后一次）：

文件"/用户/tomlous/开发/Python/丰富/构建/main.py"，第53行，job_module=importlib.import_module（'spark.jobs.%s'%args.job_name）...

文件“”，第978行，在_gcd_import中

文件""，第961行，_find_and_load

文件“”，第950行，在“查找”和“加载”中解锁

文件“”，第646行，在“加载”中

文件“”，第616行，可向后加载

文件"dependencies.zip/spark/jobs/classify_existence.py"，第9行，在

文件“dependencies.zip/enrich/existence.py”，第3行，在

文件“dependencies.zip/pandas/init.py”，第19行，在

ImportError:缺少必需的依赖项['numpy']

看看panda的\uuuu init\uuuuuuy.py我看到了类似于\uuuuuu import\uuuuuuy（numpy）

所以我假设Numpy没有加载。

但如果我将代码更改为显式调用numpy函数，它实际上会找到numpy，但不会找到它的某些依赖项

import numpy as np
a = np.array([1, 2, 3])

代码返回

回溯（最近一次呼叫最后一次）：

文件“dependencies.zip/numpy/core/init.py”，第16行，在

无法导入多阵列名称

所以我的问题是：

如何将python3库与spark作业捆绑在一起，而不必在spark集群上安装所有可能的库？

Answer 1

如果切换到virtualenv，您可以轻松实现这一点。在这个环境中，您需要安装所有必要的需求，然后使用--archives将其压缩并传递。这是一篇描述细节的好文章：https://henning.kropponline.de/2016/09/17/running-pyspark-with-virtualenv/

Answer 2

更新：有一个有凝聚力的回购，其中包括一个非常出色地做到这一点的示例项目。你应该看看，尤其是如果我下面的例子不适合你。回购在这里：https://github.com/massmutual/sample-pyspark-application并包括在YARN：https://github.com/massmutual/sample-pyspark-application/blob/master/setup-and-submit.sh上运行的示例，该示例期望您首先导出几个环境变量。（我提供的值是特定于EMR的，所以您的值可能不同。）

export HADOOP_CONF_DIR="/etc/hadoop/conf"
export PYTHON="/usr/bin/python3"
export SPARK_HOME="/usr/lib/spark"
export PATH="$SPARK_HOME/bin:$PATH"

正如这里提到的：我似乎无法让Spark上的--py文件正常工作。有必要使用virtualenv（或者conda）之类的工具来避免遇到与Python包（如Numpy）的C库编译相关的问题依赖于基础硬件体系结构，但由于依赖项和/或任务节点中的硬链接（可能与主节点实例具有不同的硬件）而无法成功移植到群集中的其他计算机。

这里讨论了--Archips和--py文件之间的一些区别：在pyspark作业中运送和使用虚拟环境

我建议使用带有虚拟环境的文件来提供包含包依赖项的压缩文件，以避免我上面提到的一些问题。

例如，从一个Amazon弹性映射减少（EMR）集群中，当ssh进入master实例时，我能够成功地使用ssh提交从一个虚拟环境中执行测试python脚本，如下所示：

pip-3.4 freeze | egrep -v sagemaker > requirements.txt
# Above line is just in case you want to port installed packages across environments.
virtualenv -p python3 spark_env3
virtualenv -p python3 --relocatable spark_env3
source spark_env3/bin/activate
sudo pip-3.4 install -U pandas boto3 findspark jaydebeapi
# Note that the above libraries weren't required for the test script, but I'm showing how you can add additional dependencies if needed.
sudo pip-3.4 install -r requirements.txt
# The above line is just to show how you can load from a requirements file if needed.
cd spark_env3
# We must cd into the directory before we zip it for Spark to find the resources. 
zip -r ../spark_env3_inside.zip *
# Be sure to cd back out after building the zip file. 
cd ..

PYSPARK_PYTHON=./spark_env3/bin/python3 spark-submit \ 
  --conf spark.yarn.appMasterEnv.PYSPARK_PYTHON=./spark_env3/bin/python3 \
  --master yarn-cluster \
  --archives /home/hadoop/spark_env3_inside.zip#spark_env3 \
  test_spark.py

请注意，上面最后一行末尾附近的主题标签不是注释。这是一个火花提交的指令，如这里所述：使用纱线上的火花提交选项-档案上传zip文件

我正在运行的测试脚本的源代码来自本文，其中讨论了如何使用conda而不是virtualenv来运行pyspark作业：http://quasiben.github.io/blog/2016/4/15/conda-spark/

并包含test_spark.py脚本的代码：

# test_spark.py
import os
import sys
from pyspark import SparkContext
from pyspark import SparkConf

conf = SparkConf()
conf.setAppName("get-hosts")

sc = SparkContext(conf=conf)

def noop(x):
    import socket
    import sys
    return socket.gethostname() + ' '.join(sys.path) + ' '.join(os.environ)

rdd = sc.parallelize(range(1000), 100)
hosts = rdd.map(noop).distinct().collect()
print(hosts)

如@Mariusz所述，如果您想了解有关使用virtualenv执行pyspark作业的一些背景信息，本博文中有一个有用的示例：https://henning.kropponline.de/2016/09/17/running-pyspark-with-virtualenv/ （虽然它没有解释我在其他链接中澄清的一些微妙之处）。

在这里提供的回复帖子中还有一个附加示例：未在PySpark中加载Elephas：没有名为Elephas.spark_model的模块

这里还有另一个例子：https://community.hortonworks.com/articles/104947/using-virtualenv-with-pyspark.html