如何将 args 传递给 Airflow 中的 DataprocSubmitJobOperator？答案

【问题标题】：How to pass args to DataprocSubmitJobOperator in Airflow?如何将 args 传递给 Airflow 中的 DataprocSubmitJobOperator？
【发布时间】：2021-12-01 12:53:26
【问题描述】：

我有一个 Spark 作业，它将参数作为键值对并将其映射到代码中，如下所示：



    val props = Utils.mapArguments(args)

    println(props)
    val gcsFolder = props("gcsPath")
    val testId=props("test.id")
    val bUrl=props("b.url")
    ......
    ......

之前，我曾经在 bash 中将作业作为 shell 脚本提交到 dataproc 集群：

gcloud dataproc jobs submit spark --async  --project testing-1 \
--cluster test-cluster \
--region us-central1 \
--properties spark.ui.killEnabled=true,\
spark.dynamicAllocation.enabled=true,\
spark.ui.enabled=true, \
--class com.walmart.ei.testClass \
--jars gs://testing-bucket/test-1.1.0.jar \
-- master=yarn \
gcsPath=gs://testing_part/path/ \
test.id=404 \
correlation.id=AUDIT_2 \
b.url=http://testing-v1.net/test/ \

现在使用气流，我们正在尝试使用 dataproc 作业提交运算符将其提交为：


# SPARK JOB CONF
SPARK_JOB = {
    "reference": {"project_id": PROJECT_ID},
    "placement": {"cluster_name": CLUSTER_NAME},
    "spark_job": {
        "jar_file_uris": ["gs://testing-bucket/test-1.1.0.jar"],
        "main_class": "com.walmart.ei.testClass",
        "gcsPath":"gs://testing_part/path/",
        "test.id":"404",
        "correlation.id":"AUDIT_2222",
        "b.url":"http://testing-v1.net/test/",
    }
}

 spark_task = DataprocSubmitJobOperator(
        task_id="spark_task", job=SPARK_JOB, location=REGION, project_id=PROJECT_ID, gcp_conn_id=CONN_ID
    )

spark_task

但此作业失败，无法将参数传递给 Spark 作业。

在 SPARK_JOB 中传递参数的正确方法是什么？

【问题讨论】：

标签： apache-spark airflow google-cloud-dataproc

【解决方案1】：

找到了一种传递参数的方法，因为它需要一个列表，我可以通过这种方式传递它来运行它。

SPARK_JOB = {
    "reference": {"project_id": PROJECT_ID},
     "placement": {"cluster_name": CLUSTER_NAME},
     "spark_job": {
        "jar_file_uris": ["gs://testing-bucket/test-1.1.0.jar"],
        "main_class": "com.walmart.ei.testClass",
         args": [
        "gcsPath":"gs://testing_part/path/",
        "test.id":"404",
        "correlation.id":"AUDIT_2222",
        "b.url":"http://testing-v1.net/test/"
         ]
    }
}

【讨论】：

您可以接受自己的答案，这样更容易被他人发现。

【解决方案2】：

job 参数是一个字典，必须与 protubuf 消息:class:~google.cloud.dataproc_v1beta2.types.Job 的形式相同（请参阅source code）

您可以查看原始消息here。

注意到有一个PR 正在进行中，以将操作员从v1beta2 迁移到v1。

【讨论】：