如何在Python方法外传递XCom列表参数并创建并行PythonOperator

核心前提：Airflow的DAG拓扑结构在DAG文件被调度器解析加载时就会完全固定，任务运行阶段无法修改DAG结构、新增/删除任务、调整依赖关系；XCom是任务执行时才会写入的运行时数据，解析阶段不存在可读取的XCom值

对你两个问题的明确答复 #

• 关于获取XCom的ids：无法在prepare_parameters函数外部的DAG解析阶段，直接读取该任务运行时才推送的ids列表。如果强行读取历史DAG运行记录里的XCom值生成任务，会导致DAG解析逻辑依赖历史运行状态，出现调度错乱、任务丢失的问题，生产环境绝对不要这么写。
• 关于在prepare_parameters内生成任务：无法在prepare_parameters的执行逻辑里生成operations任务列表搭建依赖。prepare_parameters是任务运行时才执行的函数，这时候DAG结构早就被调度器固化，运行时对DAG结构做的任何修改都不会生效。

正确实现方式 #

根据你使用的Airflow版本二选一即可，两种方式都能实现「每个id对应独立PythonOperator并行运行」的需求。

方案1：Airflow 2.3+ 动态任务映射（官方推荐，无版本兼容问题优先选这个） #

不需要提前固定任务数量，上游任务跑完拿到ids列表后，Airflow会自动为列表里的每个id展开成独立的并行任务实例，代码如下：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def prepare_parameters(**context):
    dag_conf = context['dag_run'].conf
    ids = dag_conf['ids'] if 'ids' in dag_conf else [1,2,3]
    # 直接返回ids即可，返回值会自动写入XCom，不需要手动调用xcom_push
    return ids

def do_something(id, **context):
    # 这里写单id对应的业务逻辑
    print(f"当前处理id: {id}")

with DAG(
    dag_id="parallel_id_process",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
    catchup=False,
    default_args={'queue': 'default'}
) as dag:
    prepare_parameters_operator = PythonOperator(
        python_callable=prepare_parameters,
        task_id="prepare_parameters"
    )

    # 动态映射：自动根据上游输出的ids列表，展开为多个并行的do_something任务
    operation = PythonOperator.partial(
        python_callable=do_something
    ).expand(op_kwargs={"id": prepare_parameters_operator.output})

    # 依赖链路和你预期的完全一致
    prepare_parameters_operator >> operation >> end_tasks

动态展开的每个子任务都是独立的任务实例，有独立的task_id、执行日志、重试逻辑，完全满足你的需求，不管ids列表传进来长度是多少都能自动适配。

方案2：Airflow 2.3以前老版本兼容方案 #

如果用的版本不支持动态任务映射，只能在DAG解析阶段提前把所有可能出现的id对应的任务全部定义好，运行时通过逻辑判断跳过不需要执行的任务：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

# 提前枚举业务上所有可能出现的id值，覆盖全部场景
ALL_POSSIBLE_IDS = list(range(1, 200))

def prepare_parameters(**context):
    dag_conf = context['dag_run'].conf
    ids = dag_conf['ids'] if 'ids' in dag_conf else [1,2,3]
    context['ti'].xcom_push(key='ids', value=ids)

def do_something(id, **context):
    # 拉取上游的有效ids列表，不在列表里的id直接跳过执行
    valid_ids = context['ti'].xcom_pull(task_ids="prepare_parameters", key="ids")
    if id not in valid_ids:
        return
    # 这里写单id对应的业务逻辑
    print(f"当前处理id: {id}")

with DAG(
    dag_id="parallel_id_process_old_version",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
    catchup=False,
    default_args={'queue': 'default'}
) as dag:
    prepare_parameters_operator = PythonOperator(
        python_callable=prepare_parameters,
        task_id="prepare_parameters"
    )

    operations = []
    for id in ALL_POSSIBLE_IDS:
        operations.append(PythonOperator(
            python_callable=do_something,
            task_id=f"do_something_{id}",
            op_kwargs={"id": id}
        ))

    prepare_parameters_operator >> operations >> end_tasks

这个方案的缺点是必须提前枚举所有可能的id值，如果ids范围完全不固定就没法用，优先升级Airflow用动态任务映射方案。

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