Airflow获取XCom值传递给EMR算子spark_steps的实现方法

NameError: name 'task_instance' is not defined报错的核心原因：DAG文件是调度器启动/周期刷新时解析加载的，这个阶段不存在运行时的任务实例上下文，直接在DAG顶层代码调用task_instance.xcom_pull()，根本访问不到运行时才生成的XCom值，自然触发变量未定义错误。

方案1：直接用Jinja模板语法拉取XCom（最简便） #

Airflow内置了ti（即task_instance的简写）模板宏，可直接在支持模板渲染的算子参数中调用，不需要手动定义变量。EmrCreateJobFlowOperator的job_flow_overrides参数、EmrAddStepsOperator的steps和job_flow_id参数默认都开启了模板渲染，会递归扫描配置内所有嵌套字符串做替换，完全可以直接在配置里写XCom拉取逻辑，不需要额外写Python代码拉值。
实现步骤：

• 确认get_aws_fields任务通过return返回需要的字段值（Airflow会自动将任务返回值推入XCom，不需要手动调用xcom_push）
• 在EMR算子的配置中，需要动态替换S3路径的位置直接嵌入Jinja模板语法即可

示例代码：

from airflow.providers.amazon.aws.operators.emr import EmrCreateJobFlowOperator, EmrAddStepsOperator

# 创建EMR集群
create_cluster_task = EmrCreateJobFlowOperator(
    task_id='create_emr_cluster',
    job_flow_overrides={
        "Name": "spark-etl-cluster",
        "ReleaseLabel": "emr-6.10.0",
        "LogUri": "{{ ti.xcom_pull(task_ids='get_aws_fields')['s3_log_path'] }}",
        "Instances": {
            "Ec2SubnetId": "{{ ti.xcom_pull(task_ids='get_aws_fields')['subnet_id'] }}",
            "InstanceGroups": [
                # 固定的实例配置正常写即可
            ]
        },
        # 其余固定集群配置按需补充
    },
    aws_conn_id='aws_default',
)

# 定义Spark步骤配置，动态路径用模板占位
spark_steps = [
    {
        "Name": "run_spark_etl_job",
        "ActionOnFailure": "TERMINATE_CLUSTER",
        "HadoopJarStep": {
            "Jar": "command-runner.jar",
            "Args": [
                "spark-submit",
                "--deploy-mode", "cluster",
                "{{ ti.xcom_pull(task_ids='get_aws_fields')['spark_script_s3_path'] }}",
                "--input", "{{ ti.xcom_pull(task_ids='get_aws_fields')['input_s3_path'] }}",
                "--output", "{{ ti.xcom_pull(task_ids='get_aws_fields')['output_s3_path'] }}"
            ]
        }
    }
]

# 给集群添加Spark步骤
add_steps_task = EmrAddStepsOperator(
    task_id='add_spark_steps',
    job_flow_id="{{ ti.xcom_pull(task_ids='create_emr_cluster', key='return_value') }}",
    steps=spark_steps,
    aws_conn_id='aws_default',
)

# 配置依赖顺序，必须保证get_aws_fields先执行
get_aws_fields >> create_cluster_task >> add_steps_task

方案2：复杂配置用PythonOperator动态生成后传值 #

如果Spark步骤或集群配置的拼接逻辑复杂，模板写起来可读性差，可以单独写Python任务在运行时拉取XCom、拼接完整配置，再把生成好的配置推入XCom供EMR算子读取。
示例代码：

from airflow.operators.python import PythonOperator

def generate_full_config(**context):
    # 任务执行阶段上下文存在，可正常拉取XCom
    aws_fields = context['ti'].xcom_pull(task_ids='get_aws_fields')
    # 写复杂配置拼接逻辑
    job_flow_overrides = {
        "Name": "spark-etl-cluster",
        "ReleaseLabel": "emr-6.10.0",
        "LogUri": aws_fields['s3_log_path'],
        "Instances": {
            "Ec2SubnetId": aws_fields['subnet_id'],
            # 其余实例配置
        }
    }
    spark_steps = [
        {
            "Name": "run_spark_etl_job",
            "ActionOnFailure": "TERMINATE_CLUSTER",
            "HadoopJarStep": {
                "Jar": "command-runner.jar",
                "Args": [
                    "spark-submit",
                    "--deploy-mode", "cluster",
                    aws_fields['spark_script_s3_path'],
                    "--input", aws_fields['input_s3_path'],
                    "--output", aws_fields['output_s3_path']
                ]
            }
        }
    ]
    # 把拼接好的配置推入XCom
    context['ti'].xcom_push(key='job_flow_config', value=job_flow_overrides)
    context['ti'].xcom_push(key='spark_steps', value=spark_steps)

gen_config_task = PythonOperator(
    task_id='generate_config',
    python_callable=generate_full_config,
)

create_cluster_task = EmrCreateJobFlowOperator(
    task_id='create_emr_cluster',
    # 直接拉取Python任务生成的集群配置
    job_flow_overrides="{{ ti.xcom_pull(task_ids='generate_config', key='job_flow_config') }}",
    aws_conn_id='aws_default',
)

add_steps_task = EmrAddStepsOperator(
    task_id='add_spark_steps',
    job_flow_id="{{ ti.xcom_pull(task_ids='create_emr_cluster') }}",
    # 直接拉取Python任务生成的步骤配置
    steps="{{ ti.xcom_pull(task_ids='generate_config', key='spark_steps') }}",
    aws_conn_id='aws_default',
)

# 配置依赖
get_aws_fields >> gen_config_task >> create_cluster_task >> add_steps_task

• 禁止在DAG顶层全局作用域写任何xcom_pull逻辑，DAG解析阶段没有运行时上下文，必然报错。
• get_aws_fields返回的值必须是可JSON序列化的类型（字典、列表、字符串、数字等），XCom默认用JSON序列化存储，返回自定义对象会导致序列化失败，后续任务拉取不到值。
• 必须配置正确的任务依赖，保证get_aws_fields在两个EMR算子之前执行，否则运行时拉取XCom会得到空值。

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