在google-dataproc的Spark集群中的pyspark作业中使用外部库

Question

我有一个我通过谷歌数据交换机创建的火花簇.我希望能够使用databricks 的csv库(请参阅https://github.com/databricks/spark-csv).所以我首先测试它是这样的:

我开始与我的集群的主节点进行ssh会话,然后我输入:

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

然后它启动了一个pyspark shell,我在其中输入:

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

它奏效了.

我的下一步是使用以下命令从我的主机启动此作业:

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

但在这里它不起作用,我得到一个错误.我想因为我没有--packages com.databricks:spark-csv_2.11:1.2.0作为一个论点,但我尝试了10种不同的方式来给它,我没有管理.

我的问题是:

1. 是我键入后安装的databricks csv库 pyspark --packages com.databricks:spark-csv_2.11:1.2.0
2. 我可以在我的网上写一行job.py来导入它吗？
3. 或者我应该给我的gcloud命令导入或安装它的params？

Answer 1

简答

在参数的排序中有一些怪癖,如果它出现在参数之后,--packages则不被接受.要解决此问题,您可以在从Dataproc的CLI提交时执行以下操作:spark-submitmy_job.py

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

基本上,只需在命令中添加文件--properties spark.jars.packages=com.databricks:spark-csv_2.11:1.2.0之前.py.

答案很长

所以,这实际上是一个不同于已知缺乏支持的--jars问题gcloud beta dataproc jobs submit pyspark; 看来,如果没有Dataproc明确地将其识别--packages为特殊的-level spark-submit标志,它会尝试在应用程序参数之后传递它,以便spark-submit允许将其--packages作为应用程序参数进行传递,而不是将其正确地解析为提交级别选项.事实上,在一个SSH会话,下面也不能正常工作:

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

但是,切换参数的顺序会再次起作用,即使在这种pyspark情况下,两个顺序都有效:

# 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

因此,尽管spark-submit job.py应该是以前调用的所有内容的替代品pyspark job.py,但是解析事件的差异就--packages意味着它实际上不是100%兼容的迁移.这可能是Spark方面的后续工作.

无论如何,幸运的是有一个解决方法,因为--packages它只是Spark属性的另一个别名spark.jars.packages,而Dataproc的CLI支持属性就好了.所以你可以做到以下几点:

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

请注意,--properties必须在此之前my_job.py,否则它将作为应用程序参数而不是作为配置标志发送.希望对你有用!请注意,SSH会话中的等效项将是spark-submit --packages com.databricks:spark-csv_2.11:1.2.0 job.py.

Answer 2

除了@Dennis。

请注意，如果您需要加载多个外部软件包，则需要指定一个自定义转义字符，如下所示：

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

请注意包装清单前的^＃^。请参阅gcloud topic escaping以获取更多详细信息。