如何通过Airflow DataprocSubmitJobOperator为Spark作业设置自定义Job ID

核心实现逻辑 #

DataprocSubmitJobOperator 对应gcloud命令--id参数的配置项，嵌套在作业配置的reference.job_id字段下，直接赋值即可实现自定义Job ID的效果。 #

具体配置说明 #

• 构造提交的作业配置时，新增reference字段，内部指定job_id为你需要的自定义ID值即可，注意作业ID在对应GCP项目+区域的范围内必须全局唯一，重复ID会直接导致作业提交失败。
• 支持使用Airflow内置模板变量动态生成ID，例如结合执行时间、DAG ID等字段避免ID冲突。

代码示例 #

from airflow.providers.google.cloud.operators.dataproc import DataprocSubmitJobOperator

# 自定义作业ID示例，可根据需求调整，示例中拼接了执行日期和重试次数避免重复
CUSTOM_JOB_ID = "spark-etl-task-{{ ds_nodash }}-{{ ti.try_number }}"

spark_job = DataprocSubmitJobOperator(
    task_id="submit_spark_task",
    project_id="你的GCP项目ID",
    region="Dataproc集群所在区域",
    job={
        # 自定义Job ID配置，对应gcloud --id参数
        "reference": {"job_id": CUSTOM_JOB_ID},
        "placement": {"cluster_name": "你的Dataproc集群名称"},
        "spark_job": {
            "main_class": "你自己的Spark作业主类",
            "jar_file_uris": ["gs://你的存储桶路径/spark-job.jar"],
            "args": ["--input", "gs://输入路径", "--output", "gs://输出路径"]
        }
    },
    gcp_conn_id="你的GCP连接ID"
)

注意事项 #

• 作业ID仅支持小写字母、数字、连字符，最长长度为100字符，不符合规则的ID会触发API参数校验报错。
• 如果作业开启了重试配置，建议在ID中拼接任务重试次数变量{{ ti.try_number }}，避免重试时ID重复导致提交失败。

内容的提问来源于stack exchange，提问作者Jas Kaur