Verwenden Sie eine externe Bibliothek in pyspark Job in einem Spark-Cluster von google-dataproc

Question

Ich habe einen Funken-Cluster, den ich über Google Dataproc erstellt. Ich möchte in der Lage sein, die CSV-Bibliothek von Databricks zu verwenden (siehe https://github.com/databricks/spark-csv). Also habe ich es getestet dies zunächst wie:

ich eine SSH-Sitzung mit dem Master-Knoten meines Clusters gestartet, dann Eingang I:

pyspark --packages com.databricks:spark-csv_2.11:1.2.0 

Dann ist es eine pyspark Schale ins Leben gerufen, in dem I-Eingang:

df = sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true').load('gs:/xxxx/foo.csv') 
df.show() 

Und es hat funktioniert.

Mein nächster Schritt ist es, diese Arbeit von meiner Hauptmaschine starten Sie den Befehl:

gcloud beta dataproc jobs submit pyspark --cluster <my-dataproc-cluster> my_job.py 

Aber hier funktioniert es nicht und ich erhalte eine Fehlermeldung. Ich denke, weil ich nicht die --packages com.databricks:spark-csv_2.11:1.2.0 als Argument gegeben habe, aber ich habe versucht, 10 verschiedene Möglichkeiten, es zu geben, und ich habe es nicht geschafft.

Meine Frage lautet:

1. die databricks csv-Bibliothek installiert wurde, nachdem ich pyspark --packages com.databricks:spark-csv_2.11:1.2.0 getippt
2. kann ich eine Linie in meinem job.py schreiben, um es zu importieren?
3. oder welche Parameter sollte ich meinem gcloud-Befehl geben, um es zu importieren oder zu installieren?

Answer 1

Kurze Antwort

Es gibt Macken in Reihenfolge der Argumente, wo --packages nicht von spark-submit akzeptiert, wenn sie nach dem my_job.py Argument kommt. Zur Umgehung dieses, können Sie folgendermaßen vorgehen, wenn sie von Dataproc CLI einreichen:

gcloud beta dataproc jobs submit pyspark --cluster <my-dataproc-cluster> \ 
    --properties spark.jars.packages=com.databricks:spark-csv_2.11:1.2.0 my_job.py 

Grundsätzlich fügen Sie einfach --properties spark.jars.packages=com.databricks:spark-csv_2.11:1.2.0 vor der .py Datei in Ihrem Befehl.

Lange Antwort

Also, das ist eigentlich ein anderes Problem als der bekannten Mangel an Unterstützung für --jars in gcloud beta dataproc jobs submit pyspark; es scheint, dass ohne Dataproc, das --packages als spezielle spark-submit -level Flag explizit erkennt, es versucht, es nach die Anwendungsargumente zu übergeben, damit Spark-submit das --packages durch ein Anwendungsargument fallen lässt, anstatt es ordnungsgemäß als ein submission- Level-Option.Tatsächlich in einer SSH-Sitzung, hat die folgende nicht Arbeit:

# Doesn't work if job.py depends on that package. 
spark-submit job.py --packages com.databricks:spark-csv_2.11:1.2.0 

Aber die Reihenfolge der Argumente Schalen wieder funktioniert, wenn auch im pyspark Fall beide Ordnungen arbeiten:

# Works with dependencies on that package. 
spark-submit --packages com.databricks:spark-csv_2.11:1.2.0 job.py 
pyspark job.py --packages com.databricks:spark-csv_2.11:1.2.0 
pyspark --packages com.databricks:spark-csv_2.11:1.2.0 job.py 

Obwohl spark-submit job.py eigentlich ein Ersatz für alles sein soll, was zuvor als pyspark job.py bezeichnet wurde, bedeutet der Unterschied bei der Syntaxanalyse für Dinge wie --packages, dass es sich nicht wirklich um eine 100% kompatible Migration handelt. Dies könnte etwas auf der Seite von Spark sein.

Zum Glück gibt es einen Workaround, da --packages ist nur ein weiterer Alias ​​für die Spark-Eigenschaft spark.jars.packages, und Dataprocs CLI unterstützt Eigenschaften gut. So können Sie einfach wie folgt vor:

gcloud beta dataproc jobs submit pyspark --cluster <my-dataproc-cluster> \ 
    --properties spark.jars.packages=com.databricks:spark-csv_2.11:1.2.0 my_job.py 

Beachten Sie, dass die --propertiesvor die my_job.py kommen müssen, sonst es wird als Anwendung Argument gesendet und nicht als eine Konfiguration Flagge. Hoffe das funktioniert für dich! Beachten Sie, dass das Äquivalent in einer SSH-Sitzung spark-submit --packages com.databricks:spark-csv_2.11:1.2.0 job.py wäre.

Answer 2

Zusätzlich zu @Dennis.

Beachten Sie, dass, wenn Sie mehrere externe Pakete müssen laden, müssen Sie wie so eine benutzerdefinierte Escape-Zeichen angeben:

--properties ^#^spark.jars.packages=org.elasticsearch:elasticsearch-spark_2.10:2.3.2,com.data‌​bricks:spark-avro_2.10:2.0.1 

Notiere die^#^kurz vor der Paketliste. Weitere Informationen finden Sie unter gcloud topic escaping.