3

我正在使用 Google dataproc 提交 Spark 作业并使用 Google Cloud Composer 来安排它们。不幸的是，我面临着困难。

我依靠.conf文件（类型安全配置文件）将参数传递给我的 spark 作业。

我正在为气流数据处理使用以下 python 代码：

t3 = dataproc_operator.DataProcSparkOperator(
   task_id ='execute_spark_job_cluster_test',
   dataproc_spark_jars='gs://snapshots/jars/pubsub-assembly-0.1.14-SNAPSHOT.jar',
   cluster_name='cluster',
   main_class = 'com.organ.ingestion.Main',
   project_id='project',
   dataproc_spark_properties={'spark.driver.extraJavaOptions':'gs://file-dev/fileConf/development.conf'},
   scopes='https://www.googleapis.com/auth/cloud-platform', dag=dag)

但这不起作用，我遇到了一些错误。

谁能帮我解决这个问题？
基本上我希望能够覆盖.conf文件并将它们作为参数传递给我的DataProcSparkOperator.
我也尝试过

arguments=`'gs://file-dev/fileConf/development.conf'`:

但这没有考虑.conf参数中提到的文件。

4

1 回答 1

2

tl; dr您需要将development.conf文件转换为字典以传递给dataproc_spark_properties.

完整解释：

设置属性的主要方法有两种——集群级别和作业级别。

1) 工作级别

看起来您正在尝试将它们设置在工作级别：DataProcSparkOperator(dataproc_spark_properties={'foo': 'bar', 'foo2': 'bar2'}). 这与gcloud dataproc jobs submit spark --properties foo=bar,foo2=bar2or相同spark-submit --conf foo=bar --conf foo2=bar2。这是每个作业属性的文档。

to 的参数spark.driver.extraJavaOptions应该是您将传递给 java.util 的命令行参数。例如，-verbose:gc。

2) 集群级别

您还可以使用 设置集群级别的属性DataprocClusterCreateOperator(properties={'spark:foo': 'bar', 'spark:foo2': 'bar2'})，这与gcloud dataproc clusters create --properties spark:foo=bar,spark:foo2=bar2（文档）相同。同样，您需要使用字典。

重要的是，如果您在集群级别指定属性，则需要在它们前面加上要将属性添加到的配置文件。如果你使用spark:foo=bar，那意味着添加foo=bar到/etc/spark/conf/spark-defaults.conf. 等有类似的前缀yarn-site.xml。

3）.conf在集群级别使用您的文件

如果您不想将 .conf文件转换为字典，也可以在创建集群时/etc/spark/conf/spark-defaults.conf使用初始化操作将其附加到。

例如（这是未经测试的）：

#!/bin/bash
set -euxo pipefail

gsutil cp gs://path/to/my.conf .
cat my.conf >> /etc/spark/conf/spark-defaults.conf

请注意，您要追加而不是替换现有的配置文件，以便您只覆盖您需要的配置。

于 2018-09-16T21:15:58.270 回答