Airflow2.0中TriggerDagRunOperator触发子DAG无执行日期如何取返回值

方案1：使用TriggerDagRunOperator + 子DAG Run ID拉取XCom（保留独立dag_c的场景） #

适合需要保留dag_c作为独立DAG对外提供服务的场景，完全避开execution_date重复的问题：

• 父DAG（dag_a/dag_b）中使用TriggerDagRunOperator触发dag_c，开启wait_for_completion=True参数等待子DAG运行结束，该Operator会自动将触发产生的子DAG的dag_run_id推送到父DAG的XCom中
• 父DAG后续拉取子DAG XCom时，直接指定run_id参数即可，不需要依赖execution_date，示例代码片段：

# 父DAG中触发子DAG的任务
trigger_c = TriggerDagRunOperator(
    task_id="trigger_dag_c",
    trigger_dag_id="dag_c",
    conf={"param": "your_custom_param"},
    wait_for_completion=True,
    poke_interval=30
)

# 后续拉取子DAG返回值的任务
pull_c_result = PythonOperator(
    task_id="pull_c_result",
    python_callable=lambda **context: context["ti"].xcom_pull(
        dag_id="dag_c",
        task_ids="task_that_push_result_in_c",
        run_id=context["ti"].xcom_pull(task_ids="trigger_dag_c")
    )
)

• 子DAG不需要做额外改造，保持原有xcom_push逻辑即可。

方案2：封装公共逻辑为TaskGroup（官方推荐替代SubDAG的方案） #

该方案是Airflow官方推荐的SubDAG替代方案，既能复用dag_c的逻辑，又不需要维护独立的子DAG，完全不存在跨DAG拉取XCom的问题：

• 将dag_c的所有业务逻辑封装为一个公共的TaskGroup工厂函数，存放在公共工具模块中
• 在dag_a和dag_b中直接导入该工厂函数，生成对应场景的任务组，所有任务组内的XCom可以直接在父DAG中读取，不需要跨DAG操作
  示例代码片段：

# 公共模块 common.py 中封装dag_c逻辑
from airflow.utils.task_group import TaskGroup

def create_dag_c_taskgroup(group_id, input_param, **kwargs):
    with TaskGroup(group_id=group_id, **kwargs) as tg:
        # 原dag_c的所有任务逻辑直接写在这里，input_param是自定义入参
        task1 = PythonOperator(...)
        task2 = PythonOperator(...)
        task1 >> task2
    return tg

# dag_a/dag_b中直接调用
from common import create_dag_c_taskgroup

with DAG(dag_id="dag_a", ...) as dag:
    upstream_task = PythonOperator(...)
    # 生成专属的dag_c逻辑任务组
    dag_c_tg = create_dag_c_taskgroup(
        group_id="dag_c_for_a",
        input_param="a的专属参数"
    )
    downstream_task = PythonOperator(
        task_id="use_c_result",
        python_callable=lambda **context: context["ti"].xcom_pull(
            task_ids="dag_c_for_a.task_that_push_result"
        )
    )
    upstream_task >> dag_c_tg >> downstream_task

方案选型建议 #

如果dag_c需要支持独立触发运行，选择方案1；如果dag_c仅为dag_a和dag_b提供公共逻辑，选择方案2，后者运维成本更低，无跨DAG通信风险。

内容的提问来源于stack exchange，提问作者Sachin M Naik