如何通过Airflow的DataprocCreateBatchOperator指定Dataproc版本？

要解决Dataproc默认版本升级导致任务失败的问题，你可以通过在batch配置的environment_config中添加runtime_config字段来指定具体的Dataproc版本，这对应gcloud命令中的--version参数。

修改你的任务代码，在environment_config内加入runtime_config配置：

task2 = DataprocCreateBatchOperator(
    task_id="trip_level_data",
    project_id="generic_project_id",
    region="us-east4",
    batch_id="trip-"+"".join(random.choice(string.ascii_lowercase + string.digits) for i in range(35)),
    batch={
        "pyspark_batch" : {
            "args" : [
                "--env=prod"
            ],
            "jar_file_uris" : [
                "gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.23.0.jar"
            ],
            "main_python_file_uri" : "gs://generic_bucket/get_trip_info.py"
        },
        "labels" : {
            "dag_id" : "{{ run_id_format(dag.dag_id) }}",
            "dag_run_id" : "{{ run_id_format(run_id) }}",
            "task_id" : "{{ run_id_format(task.task_id) }}"
        },
        "environment_config" : {
            "execution_config" : {
                "service_account" : "svc@generic_project.iam.gserviceaccount.com",
                "subnetwork_uri" : "https://www.googleapis.com/compute/alpha/projects/shared-vpc-admin/regions/us-east4/subnetworks/prod-us-east4-01"
            },
            # 添加runtime_config指定Dataproc版本
            "runtime_config": {
                "version": "2.0.85"
            }
        }
    }
)

说明 #

runtime_config.version参数直接对应Dataproc Batch API中的版本指定，和gcloud命令的--version参数功能完全一致。通过这个配置，Airflow提交的Batch任务会使用你指定的Dataproc版本运行，避免默认版本升级带来的兼容性问题。

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