Airflow 2.x.x动态创建TriggerDagRunOperator及相关问题咨询

我有一个包含两个组件的父DAG：

• task_a：调用外部REST API获取子DAG详情（DAG ID和参数）
• task_b：基于task_a的响应触发对应子DAG

示例task_a响应：

[
    {'id': 1, 'dag_id': 'child_1', 'params': {'test': 'test1'}},
    {'id': 2, 'dag_id': 'child_2', 'params': {'test': 'test2'}},
    {'id': 3, 'dag_id': 'child_3', 'params': {'test': 'test3'}}
]

我编写了如下DAG代码尝试创建TriggerDagRunOperator，但无法触发子DAG：

import logging
import sys

import airflow
from airflow.utils.dates import days_ago
from airflow import DAG
from airflow.models.variable import Variable
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
from airflow.operators.python import PythonOperator
from airflow.decorators import task
import requests
from airflow.operators.bash import BashOperator
from airflow.decorators import task, task_group

# from airflow.decorators import dag, dag_run_trigger

doc_md = """
## Purpose
This DAG will fetch pipeline details from pipeline master table and trigger downstream pipelines dynamically.
"""

default_args = {
    'owner': 'Airflow',
    'start_date': days_ago(1),
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1
}

with DAG(
    dag_id="core_forecasting_automated_ml_pipeline",
    default_args=default_args,
    schedule_interval=None,
    catchup=False,
    doc_md=doc_md
) as dag:

    @task
    def task_a():
        api_url = 'https://<external_host>/get_pipelines'
        response = requests.get(api_url)
        return response.json()

    @task
    def task_b(dag_info):
        child_dag_trigger = TriggerDagRunOperator(
            task_id=f'trigger_sub_dags',
            trigger_dag_id=dag_info['dag_id'],
            conf=dag_info['params'],
            wait_for_completion=True,
            poke_interval=20,
            allowed_states=['success'],
            dag=dag
        )
        child_dag_trigger
 

    dag_info = task_a()
    dag_info >> task_b.expand(dag_info=dag_info)

Airflow版本：v2.5.3

我的问题：

1. 如何动态映射TriggerDagRunOperator，并传入trigger_dag_id和conf参数？
2. TriggerDagRunOperator是否有等效装饰器？

更新：尝试过程中遇到的错误 #

错误1：动态映射时提示TypeError

Broken DAG: [/usr/local/airflow/dags/automated_ml.py] Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/airflow/decorators/task_group.py", line 97, in _create_task_group
    retval = self.function(*args, **kwargs)
  File "/usr/local/airflow/dags/automated_ml.py", line 95, in dynamic_tg
    trigger_dag_id= dag_info['dag_id'] if not dag_info['dag_id'] else '',
TypeError: 'MappedArgument' object is not subscriptable

错误2：使用动态任务映射时，无法在运行时传入trigger_dag_id

尝试代码：

TriggerDagRunOperator.partial(
        task_id='triger_child_dags',
        wait_for_completion=True,
        poke_interval=20,
        allowed_states=['success'],
    ).expand(
        trigger_dag_id=pipeline_info['dag_id'], # 无法传入字符串值，Airflow期望列表或字典类型
        conf=pipeline_info,
    )

问题1：动态映射TriggerDagRunOperator的正确方式 #

在Airflow 2.x的动态任务映射中，TriggerDagRunOperator的trigger_dag_id属于模板化字段，但直接用.expand()映射该字段会因为DAG解析阶段要求静态值而报错。以下是两种可行的实现方案：

方案1：在Python任务中调用Airflow内部API触发子DAG #

将task_b改为Python任务，通过Airflow的本地客户端API触发子DAG，可完全动态传递dag_id和参数，还能实现等待子DAG完成的逻辑：

@task
def task_b(dag_info):
    from airflow.api.client.local_client import Client
    from airflow.utils.state import State

    client = Client(None, None)
    # 触发子DAG并获取运行ID
    run_response = client.trigger_dag(
        dag_id=dag_info['dag_id'],
        conf=dag_info['params']
    )
    run_id = run_response["run_id"]

    # 轮询等待子DAG完成
    while True:
        dag_run = client.get_dag_run(dag_id=dag_info['dag_id'], run_id=run_id)
        if dag_run.state in [State.SUCCESS, State.FAILED, State.SKIPPED]:
            break
        time.sleep(20)
    
    # 子DAG失败则抛出异常终止父任务
    if dag_run.state != State.SUCCESS:
        raise Exception(f"子DAG {dag_info['dag_id']} 执行失败，状态：{dag_run.state}")

保持动态映射逻辑：

dag_info = task_a()
dag_info >> task_b.expand(dag_info=dag_info)

方案2：使用动态任务组结合模板变量 #

利用@task_group和动态映射，在任务组内部创建TriggerDagRunOperator，通过XCom模板变量传递动态参数：

@task_group
def trigger_child_dags(dag_idx):
    # 从task_a的返回结果中按索引获取对应子DAG信息
    TriggerDagRunOperator(
        task_id=f"trigger_child_{dag_idx}",
        trigger_dag_id="{{ ti.xcom_pull(task_ids='task_a')[dag_idx]['dag_id'] }}",
        conf="{{ ti.xcom_pull(task_ids='task_a')[dag_idx]['params'] }}",
        wait_for_completion=True,
        poke_interval=20,
        allowed_states=['success']
    )

dag_info = task_a()
# 基于task_a返回的列表长度动态生成任务组实例
trigger_child_dags.expand(dag_idx=range(len(dag_info)))

问题2：TriggerDagRunOperator的等效装饰器 #

Airflow 2.5.3版本没有直接对应TriggerDagRunOperator的官方装饰器，但可以通过以下方式实现类似效果：

1. 采用方案1的Python任务写法，本质就是装饰器风格的触发逻辑
2. 自定义装饰器封装Airflow API调用逻辑

从Airflow 2.6+版本开始，官方新增了@dag_run_trigger装饰器，但你的版本无法直接使用。如果无法升级Airflow，建议优先使用方案1的实现方式。

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