Airflow 2.0.2(MWAA)如何不使用Operator直接拉取XCom内的数据

核心问题说明 #

你的需求本质是混淆了Airflow的两个完全独立的运行阶段：

1. DAG解析阶段：调度器周期性加载DAG代码文件的阶段，此时你写的所有顶层Python代码（包括正则提取s3Path、构造sparkstep、初始化EmrAddStepsOperator的逻辑）都会直接执行，这个阶段没有任何DAG运行实例，也没有任务实例（ti），不可能拿到只有运行时才产生的XCom数据
2. DAG运行阶段：DAG被触发后，任务逐个调度执行的阶段，此时才会生成ti对象，也才能读写XCom

不存在不通过Operator在DAG解析阶段拿到ti操作XCom的方法，所有运行时动态值的传递必须依赖Airflow的模板能力或者任务执行逻辑内的操作。

解决方案 #

方案1：使用Jinja模板直接构造参数（无需额外Operator） #

EmrAddStepsOperator的steps参数默认支持Jinja模板渲染，你可以直接在参数中嵌入XCom拉取和处理逻辑，Airflow会在该任务执行前自动渲染出实际值：

1. 如果你的正则逻辑比较简单，可以直接用Airflow内置的regex_search过滤器：

SQSRUN = SQSSensor(
    task_id='sqs',
    poke_interval=0,
    timeout=10,
    sqs_queue=SQS_URL,
    aws_conn_id=AWS
)

sparkstep = [
    {
        "Name": "Spark转换任务",
        "ActionOnFailure": "CONTINUE",
        "HadoopJarStep": {
            "Jar": "command-runner.jar",
            "Args": [
                "spark-submit",
                "--class", "你的主类全限定名",
                "s3://你的Spark jar包路径",
                "{{ ti.xcom_pull(task_ids='sqs', key='messages')['Messages'][0]['Body'] | regex_search('para1=\\w+') }}",
                "{{ ti.xcom_pull(task_ids='sqs', key='messages')['Messages'][0]['Body'] | regex_search('para2=\\w+') }}"
            ]
        }
    }
]

sparkTransform = EmrAddStepsOperator(
    task_id='S3PathTransform',
    job_flow_id=Variable.get("EMR"),
    aws_conn_id=AWS,
    steps=sparkstep
)

# 任务依赖直接关联即可
SQSRUN >> sparkTransform

2. 如果正则逻辑比较复杂，可以自定义Jinja过滤器挂载到DAG：

# 自定义参数提取函数
def extract_para(s3_body, para_name):
    import re
    match_res = re.findall(rf"({para_name}=\w+)", s3_body)
    return match_res[0] if match_res else ""

# 将过滤器挂载到DAG实例
dag = DAG(
    # 你的DAG原有配置
)
dag.user_defined_filters = {"extract_para": extract_para}

# 模板中直接调用自定义过滤器
sparkstep = [
    {
        # 其他配置不变
        "Args": [
            "spark-submit",
            "--class", "你的主类全限定名",
            "s3://你的Spark jar包路径",
            "{{ ti.xcom_pull(task_ids='sqs', key='messages')['Messages'][0]['Body'] | extract_para('para1') }}",
            "{{ ti.xcom_pull(task_ids='sqs', key='messages')['Messages'][0]['Body'] | extract_para('para2') }}"
        ]
    }
]

方案2：使用TaskFlow API处理逻辑（适合复杂转换场景） #

如果你需要对s3Path做大量处理，可以用Airflow2.0+支持的TaskFlow API，把处理逻辑封装成任务，输出直接传递给下游Operator：

from airflow.decorators import task
import re

@task
def process_s3_path(**context):
    msg = context['ti'].xcom_pull(task_ids='sqs', key='messages')
    s3_path = msg['Messages'][0]['Body']
    para1 = re.findall(r"(para1=\w+)", s3_path)[0]
    para2 = re.findall(r"(para2=\w+)", s3_path)[0]
    # 直接构造完整的sparkstep返回
    return [
        {
            "Name": "Spark转换任务",
            "ActionOnFailure": "CONTINUE",
            "HadoopJarStep": {
                "Jar": "command-runner.jar",
                "Args": [
                    "spark-submit",
                    "--class", "你的主类全限定名",
                    "s3://你的Spark jar包路径",
                    para1,
                    para2
                ]
            }
        }
    ]

SQSRUN = SQSSensor(
    task_id='sqs',
    poke_interval=0,
    timeout=10,
    sqs_queue=SQS_URL,
    aws_conn_id=AWS
)

sparkTransform = EmrAddStepsOperator(
    task_id='S3PathTransform',
    job_flow_id=Variable.get("EMR"),
    aws_conn_id=AWS,
    steps="{{ ti.xcom_pull(task_ids='process_s3_path') }}"
)

# 任务依赖
SQSRUN >> process_s3_path() >> sparkTransform

其他错误说明 #

你之前尝试把s3Path存为Airflow Variable的方案不可行：Airflow Variable是全局静态配置，会在DAG解析阶段就被读取，而s3Path是每次DAG运行动态生成的运行时数据，不仅会有并发覆盖的问题，也不可能在解析阶段拿到运行时才写入的值。

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