如何正确配置DataprocCreateClusterOperator的暂存与临时存储桶？

你直接在DataprocCreateClusterOperator中设置staging_bucket和temp_bucket是错误的，因为该Operator本身没有这两个顶层参数，直接设置会被忽略甚至引发参数错误。正确的做法是将这两个存储桶配置为集群的Spark/Hadoop属性，添加到cluster_config的software_config.properties中。

正确配置逻辑 #

需要在CLUSTER_CONFIG的software_config.properties里添加对应属性：

• 暂存存储桶：通过spark:spark.hadoop.fs.staging.dir指定GCS路径（格式为gs://bucket/path）
• 临时存储桶：通过spark:spark.hadoop.fs.temp.dir指定GCS路径

修改后的完整代码 #

import pendulum
from datetime import timedelta

from airflow.providers.google.cloud.operators.dataproc import DataprocCreateClusterOperator
from airflow.decorators import dag

config = {
    "cluster_name": 'teste',
    "region": 'us-central1',
    "project_id": 'test-dataproc'
}

default_args = {
    "owner": "Airflow",
    "start_date": pendulum.datetime(2024, 4, 26, tz="America/Sao_Paulo"),
    "retries": 0,
    'retry_delay': timedelta(minutes=3)
}

@dag(
    "teste_dataproc",
    default_args=default_args,
    schedule_interval='0 3 * * 1-5',
    params=config,
    catchup=False,
    tags=["TESTE"],
)
def teste_dataproc():

    CLUSTER_CONFIG = {
        "gce_cluster_config" : {
            "zone_uri": "us-central1-a",
            "internal_ip_only": False,
        },
        "master_config": {
            "num_instances": 1,
            "machine_type_uri": "n1-standard-2",
            "disk_config": {"boot_disk_type": "pd-standard", "boot_disk_size_gb": 25},
        },
        "worker_config": {
            "num_instances": 2,
            "machine_type_uri": "n1-standard-2",
            "disk_config": {"boot_disk_type": "pd-standard", "boot_disk_size_gb": 25},
        },
        "software_config": {
            "properties": {
                "spark:spark.executor.memory": "3g",
                "spark:spark.executor.cores": "2",
                "spark:spark.executor.instances": "4",
                # 新增Staging与Temp存储桶配置
                "spark:spark.hadoop.fs.staging.dir": "gs://teste/dataproc/staging",
                "spark:spark.hadoop.fs.temp.dir": "gs://teste/dataproc/temp"
            }
        }
    }

    create_cluster = DataprocCreateClusterOperator(
        task_id="create_dataproc_cluster",
        project_id="{{ params.project_id }}",
        cluster_config=CLUSTER_CONFIG,
        region="{{ params.region }}",
        cluster_name="{{ params.cluster_name }}"
    )

    create_cluster

teste_dataproc()

额外注意事项 #

• 确保Airflow使用的服务账号（或Dataproc集群默认服务账号）拥有对应GCS桶的读写权限，否则集群运行时会出现权限报错。
• 如果需要为Dataproc集群自身操作（如作业提交临时文件）指定存储桶，可额外添加dataproc:dataproc.temp.bucket属性，值为GCS桶名称（不带gs://前缀）。

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