Dagster自动化构建EMR集群后PySpark无法在EMR运行问题

问题出在emr_pyspark_step_launcher资源未正确绑定到刚启动的EMR集群ID。硬编码集群ID时能正常运行，说明资源配置本身无问题，但静态获取集群ID的逻辑没把值动态注入到资源中，导致PySpark步骤默认使用本地环境执行。

• 放弃在资源初始化时静态获取集群ID，改为通过启动集群的op输出结果动态传递集群ID给emr_pyspark_step_launcher
• 利用Dagster的configured方法，在Job层面动态绑定集群ID到资源，确保步骤提交时拿到的是刚启动的集群ID
• 强制集群启动后进入RUNNING状态再提交步骤，避免因集群未就绪导致的异常

from dagster import op, job, resource
from dagster_aws.emr import EmrJobRunner, emr_pyspark_step_launcher
from dagster_pyspark import pyspark_resource

# 替换为你的EMR集群配置
EMR_CLUSTER_CONFIG = {
    "Name": "dagster-auto-emr-cluster",
    "ReleaseLabel": "emr-6.10.0",
    "Instances": {
        "InstanceGroups": [
            {
                "Name": "Master",
                "Market": "ON_DEMAND",
                "InstanceRole": "MASTER",
                "InstanceType": "m5.xlarge",
                "InstanceCount": 1,
            },
            {
                "Name": "Core",
                "Market": "ON_DEMAND",
                "InstanceRole": "CORE",
                "InstanceType": "m5.xlarge",
                "InstanceCount": 2,
            },
        ],
        "KeepJobFlowAliveWhenNoSteps": True,
        "TerminationProtected": False,
    },
    "JobFlowRole": "EMR_EC2_DefaultRole",
    "ServiceRole": "EMR_DefaultRole",
}

@op
def start_emr_cluster():
    runner = EmrJobRunner(region="us-east-1")
    # 启动集群
    cluster_id = runner.run_job_flow(EMR_CLUSTER_CONFIG)
    # 等待集群完全就绪，否则提交步骤会失败
    runner.wait_for_cluster_running(cluster_id)
    return cluster_id

@op
def terminate_emr_cluster(cluster_id: str):
    runner = EmrJobRunner(region="us-east-1")
    runner.terminate_job_flow(cluster_id)

# 定义可动态接收集群ID的资源
@resource(config_schema={"cluster_id": str, "region": str})
def dynamic_emr_step_launcher(context):
    return emr_pyspark_step_launcher.configured(
        {
            "cluster_id": context.resource_config["cluster_id"],
            "region": context.resource_config["region"],
        }
    )()

@op(required_resource_keys={"emr_pyspark_step_launcher", "pyspark"})
def run_pyspark_processing(context):
    # 替换为你的实际PySpark业务逻辑
    spark = context.resources.pyspark.spark_session
    df = spark.read.text("s3://your-input-bucket/raw-data/")
    word_count_df = df.selectExpr("explode(split(value, ' ')) as word").groupBy("word").count()
    word_count_df.write.mode("overwrite").parquet("s3://your-output-bucket/processed-data/")

@job(resource_defs={"pyspark": pyspark_resource})
def automated_emr_pipeline():
    cluster_id = start_emr_cluster()
    # 动态绑定刚启动的集群ID到资源
    step_launcher = dynamic_emr_step_launcher.configured(
        {"cluster_id": cluster_id, "region": "us-east-1"}
    )
    # 用绑定后的资源执行PySpark步骤
    run_pyspark_processing.with_resources({"emr_pyspark_step_launcher": step_launcher})()
    # 所有步骤完成后终止集群
    terminate_emr_cluster(cluster_id)

• 集群就绪等待：必须调用wait_for_cluster_running，否则集群处于初始化状态时，提交的步骤可能会 fallback 到本地执行
• 动态资源绑定：在Job中通过configured方法将op输出的集群ID注入资源，确保资源拿到的是当前流水线启动的集群ID，而非静态配置的旧值
• 依赖顺序：PySpark步骤op必须依赖启动集群的op输出，Dagster会自动保证集群启动完成后再执行数据处理步骤

内容的提问来源于stack exchange，提问作者D. Gal