我正在尝试运行依赖于某些 python3 库的 PySpark 作业。
我知道我可以在 Spark 集群上安装这些库,但由于我正在为多个作业重用该集群,所以我宁愿捆绑所有依赖项并通过 --py-files< 将它们传递给每个作业
指令。
为此,我使用:
pip3 install -r requirements.txt --target ./build/dependencies
cd ./build/dependencies
zip -qrm . ../dependencies.zip
它有效地压缩了所需包中的所有代码,以便在根级别使用。
在我的 main.py
中,我可以导入依赖项
if os.path.exists('dependencies.zip'):
sys.path.insert(0, 'dependencies.zip')
并将 .zip 添加到我的 Spark 上下文中
sc.addPyFile('dependencies.zip')
到目前为止一切顺利。
但出于某种原因,这将在 Spark 集群上产生某种依赖 hell
例如运行
spark-submit --py-files dependencies.zip main.py
我想在 main.py
(或类)中使用 Pandas 。它将触发此错误的代码:
Traceback (most recent call last):
File "/Users/tomlous/Development/Python/enrichers/build/main.py", line 53, in job_module = importlib.import_module('spark.jobs.%s' % args.job_name) ...
File "", line 978, in _gcd_import
File "", line 961, in _find_and_load
File "", line 950, in _find_and_load_unlocked
File "", line 646, in _load_unlocked
File "", line 616, in _load_backward_compatible
File "dependencies.zip/spark/jobs/classify_existence.py", line 9, in
File "dependencies.zip/enrich/existence.py", line 3, in
File "dependencies.zip/pandas/init.py", line 19, in
ImportError: Missing required dependencies ['numpy']
看 Pandas 的 __init__.py
我看到类似 __import__(numpy)
所以我假设 numpy 没有加载。
但是如果我更改我的代码以显式调用 numpy 函数,它实际上会找到 numpy,但不是它的某些依赖项
import numpy as np
a = np.array([1, 2, 3])
代码返回
Traceback (most recent call last):
File "dependencies.zip/numpy/core/init.py", line 16, in
ImportError: cannot import name 'multiarray'
所以我的问题是:
我应该如何将 python3 库与我的 spark 作业捆绑在一起,而不用 pip3 在 Spark 集群上安装所有可能的库?
最佳答案
更新:有一个完整的存储库,其中包含一个非常出色地完成此任务的示例项目。你应该看看,尤其是如果我下面的例子不适合你的话。 repo 协议(protocol)在这里: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"
如此处所述:I can't seem to get --py-files on Spark to work 有必要使用类似 virtualenv 的东西(或者 conda 可能工作)来避免遇到与 Python 包(例如 Numpy)的 C 库编译相关的问题,这些问题依赖于底层硬件架构,无法成功移植到由于依赖关系和/或任务节点中的硬链接(hard link),集群中的其他机器可能具有与主节点实例不同的硬件。
这里讨论了 --archives 和 --py-files 之间的一些区别:Shipping and using virtualenv in a pyspark job
我建议使用 --archives 和 virtualenv 来提供包含包依赖项的压缩文件,以避免我上面提到的一些问题。
例如,在 Amazon Elastic Map Reduce (EMR) 集群中,当通过 ssh 连接到主实例时,我能够成功地使用 spark-submit 从 virtualenv 环境中执行测试 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
请注意,上面最后一行末尾附近的主题标签不是评论。这是一个 spark-submit 的指令,如此处解释:Upload zip file using --archives option of spark-submit on yarn
我正在运行的测试脚本的来源来自这篇讨论使用 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)
如果您想了解有关使用 virtualenv 执行 pyspark 作业的一些背景信息,正如@Mariusz 已经提到的,这篇博文中有一个有用的示例:https://henning.kropponline.de/2016/09/17/running-pyspark-with-virtualenv/ (虽然它没有解释我用我提供的其他链接澄清的一些微妙之处)。
此处提供的答案帖子中还有一个附加示例:Elephas not loaded in PySpark: No module named elephas.spark_model
这里还有另一个例子:https://community.hortonworks.com/articles/104947/using-virtualenv-with-pyspark.html
关于python - 为 PySpark 捆绑 Python3 包导致缺少导入,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/48770263/