在 google cloud dataproc 上使用 spark 管理 python 依赖项

Question

我编写了一个 Spark 脚本，它依赖于 six 和其他各种 python 包。

$ cat ./test_package/__init__.py
from six.moves.urllib.request import urlopen

def download_size(url):
    return len(urlopen(url).read())

因此，我写了一个 setup.py 来说明这些依赖关系。

$ cat ./setup.py
from setuptools import setup
setup(
    name="Test App",
    packages=['test_package'],
    version="0.1",
    install_requires=['six>=1.0'],

然后在我的 spark 脚本中，我有需要包的代码

$ cat spark_script.py
#!/usr/lib/spark/bin/spark-submit
from pyspark import SparkContext
from glob import glob
from test_package import download_size

sc = SparkContext()
sc.addPyFile(glob('dist/Test_App-*.egg')[0])

...

sc.parallelize(urls).map(download_size).collect()

如果我跑步

$ ./test.py

它工作正常。但是，如果我尝试使用 python3，

$ PYSPARK_PYTHON=python3 ./test.py

主节点能够导入test_package，但是在mapreduce的中间我在每个工作节点上都得到了这个：

  File "/hadoop/yarn/nm-local-dir/usercache/sam/appcache/application_1487279780844_0041/container_1487279780844_0041_01_000003/pyspark.zip/pyspark/serializers.py", line 419, in loads
    return pickle.loads(obj, encoding=encoding)
  File "./Test_App-0.1-py2.7.egg/test_package/__init__.py", line 2, in <module>
    from six.moves.urllib.request import urlopen
ImportError: No module named 'six'

如何在 Google Cloud Dataproc 配置的 Apache Spark 集群上管理 Python 依赖项？

Answer 1

由于工作程序任务将在工作程序节点上运行，并且您仅手动安装了额外的 python 包，因此工作程序节点没有与您的主节点相同的可用配置。

您应该使用Dataproc initialization actions 在集群部署时在集群的所有节点上运行您的自定义脚本。对于像PYSPARK_PYTHON 这样的环境变量，您可能需要将这些设置附加到/etc/spark/conf/spark-env.sh。