Cloud Composer调度Dataproc PySpark作业时遇ValueError问题求助

核心问题 #

1. 类属性初始化时机过早：MyClass的bucket是类级静态属性，会在Python解释器加载类时就执行初始化。此时Spark的driver/executor进程可能还未读取到DATA_BUCKET_NAME环境变量，导致GCS_BUCKET_NAME为None，进而触发ValueError: Cannot determine path without bucket name。
2. 环境变量未传递到Spark进程：初始化脚本把环境变量写入/etc/environment，但Spark的driver和executor进程默认不会读取这个文件，只有SSH登录的shell进程会加载它。
3. 代码缩进错误：原代码中blob.upload_from_string(contents)未缩进在load_data方法内，属于逻辑错误。

修复步骤 #

1. 调整MyClass的bucket初始化逻辑 #

将bucket的初始化从类属性移到实例的__init__方法中，确保每次实例化类时才获取bucket，此时环境变量已生效：

from google.cloud import storage
from .constants import GCS_BUCKET_NAME

class MyClass:
    def __init__(self):
        # 实例化时才初始化bucket，确保GCS_BUCKET_NAME已加载
        self.bucket = storage.Client().bucket(GCS_BUCKET_NAME)

    def load_data(self, filepath: str, contents: str) -> None:
        blob = self.bucket.blob(filepath)
        blob.upload_from_string(contents)  # 修复缩进，确保在方法内执行

2. 确保Spark进程能获取环境变量 #

两种可选方式：

方式A：在Cloud Composer提交作业时传递环境变量

通过DataprocSubmitJobOperator的properties参数，直接将环境变量传递给Spark的driver和executor：

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

submit_job = DataprocSubmitJobOperator(
    task_id="run_pyspark_job",
    project_id="your-project-id",
    region="your-region",
    job={
        "reference": {"project_id": "your-project-id"},
        "placement": {"cluster_name": "your-dataproc-cluster"},
        "pyspark_job": {
            "main_python_file_uri": "gs://your-bucket/path/to/your/job.py",
            "properties": {
                # 传递环境变量给driver和executor
                "spark.driverEnv.DATA_BUCKET_NAME": "your-bucket-name",
                "spark.executorEnv.DATA_BUCKET_NAME": "your-bucket-name"
            }
        }
    }
)

方式B：修改Dataproc初始化脚本，注入Spark配置

修改初始化脚本，把环境变量添加到Spark的spark-env.sh中，确保Spark进程启动时自动加载：

#!/bin/bash
DATA_BUCKET_NAME=$(/usr/share/google/get_metadata_value attributes/DATA_BUCKET_NAME)
echo "DATA_BUCKET_NAME=${DATA_BUCKET_NAME}" >> /etc/environment
# 将环境变量添加到Spark环境配置
echo "export DATA_BUCKET_NAME=${DATA_BUCKET_NAME}" >> /etc/spark/conf/spark-env.sh

3. 权限验证（可选） #

再次确认Dataproc集群的服务账号拥有storage.objects.create和storage.buckets.get权限，确保能正常访问GCS bucket。

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