Verwalten von Python-Abhängigkeiten mit Spark in google cloud dataproc

Question

Ich habe ein Spark-Skript geschrieben, das von six und verschiedenen anderen Python-Paketen abhängt.

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

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

Als solche habe ich ein setup.py geschrieben, die solche Abhängigkeiten erklärt.

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

Da ist in meinem Funken Skript habe ich Code, der das Paket

$ 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() 

Wenn ich

laufen erfordert

$ ./test.py 

Es funktioniert gut. Allerdings, wenn ich versuche python3 zu verwenden,

$ PYSPARK_PYTHON=python3 ./test.py 

Der Master-Knoten ist in der Lage test_package zu importieren, aber in der Mitte des mapreduce ich dies auf jeden Arbeiter Knoten:

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' 

Wie ich tue Python-Abhängigkeiten von einem Google Cloud Dataproc-bereitgestellten Apache-Funke-Cluster verwalten?

Answer 1

Da Worker-Tasks auf Worker-Knoten ausgeführt werden und Sie nur Ihre zusätzlichen Python-Pakete manuell installieren, verfügen die Worker-Knoten nicht über die gleiche Konfiguration wie Ihr Master-Knoten.

Sie sollten Dataproc initialization actions verwenden, um Ihre Anpassungsskripts zur Clusterbereitstellungszeit auf allen Knoten des Clusters auszuführen. Für Umgebungsvariablen wie PYSPARK_PYTHON müssen Sie diese Einstellungen wahrscheinlich an /etc/spark/conf/spark-env.sh anhängen.