Airflow Controller DAG触发子DAG后未立即结束及任务内触发问题求助

核心需求 #

• Controller DAG触发子DAG后立即结束，无需等待子DAG执行完成
• 在Airflow任务运行时调用API并触发子DAG，避免DAG解析阶段执行API逻辑

方案一：动态任务映射（Airflow 2.3+ 推荐） #

该方案既避免了DAG解析时调用API，又能动态生成触发任务，且每个触发任务在完成子DAG触发后立即标记成功，Controller DAG会在所有触发任务完成后结束（无需等待子DAG）。

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
import pendulum
import requests

def fetch_api_data():
    # 仅在任务运行时调用API
    response = requests.get("YOUR_API_ENDPOINT")
    response.raise_for_status()
    # 为每个条目添加索引，方便生成唯一task_id
    return [{"index": idx, **item} for idx, item in enumerate(response.json())]

with DAG(
    dag_id="controller_dag_per",
    start_date=pendulum.datetime(2024, 4, 22, tz="UTC"),
    schedule="0/5 * * * *",
    catchup=False,
    is_paused_upon_creation=False,
    max_active_runs=100,
    max_active_tasks=50,
) as dag:
    # 任务1：调用API获取待触发的子DAG列表
    fetch_task = PythonOperator(
        task_id="fetch_subdag_list",
        python_callable=fetch_api_data
    )

    # 任务2：动态生成TriggerDagRunOperator，基于API返回的列表
    trigger_subdags = TriggerDagRunOperator.partial(
        task_id="trigger_subdag",
        reset_dag_run=True,
        wait_for_completion=False,  # 关键：触发后立即标记任务完成
        conf={}
    ).expand(
        # 为每个条目生成对应的dag_id和task_id
        trigger_dag_id=fetch_task.output.map(lambda x: f"test_dag_{x['dag_id']}_dag"),
        task_id=fetch_task.output.map(lambda x: f"{x['dag_id']}_{x['my_var']}_{x['index']}")
    )

    # 设置依赖：先获取数据，再触发子DAG
    fetch_task >> trigger_subdags

方案二：Python任务内直接触发子DAG（适合复杂逻辑场景） #

如果需要自定义触发逻辑（如条件判断、异常处理），可使用Airflow内部API在Python任务内触发子DAG，无需依赖TriggerDagRunOperator。

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.api.client.local_client import Client
import pendulum
import requests

def trigger_subdags_in_task():
    # 调用API获取数据
    response = requests.get("YOUR_API_ENDPOINT")
    response.raise_for_status()
    results_list = response.json()

    # 初始化Airflow本地客户端
    client = Client(None, None)

    for index, item in enumerate(results_list):
        try:
            target_dag_id = f"test_dag_{item['dag_id']}_dag"
            # 触发子DAG
            client.trigger_dag(
                dag_id=target_dag_id,
                conf={},
                reset_dag_run=True,
                external_trigger=True
            )
            print(f"Successfully triggered DAG: {target_dag_id}")
        except Exception as e:
            # 自定义异常处理（如日志记录、告警）
            print(f"Failed to trigger {target_dag_id}: {str(e)}")
            continue

with DAG(
    dag_id="controller_dag_per",
    start_date=pendulum.datetime(2024, 4, 22, tz="UTC"),
    schedule="0/5 * * * *",
    catchup=False,
    is_paused_upon_creation=False,
    max_active_runs=100,
    max_active_tasks=50,
) as dag:
    trigger_task = PythonOperator(
        task_id="trigger_all_subdags",
        python_callable=trigger_subdags_in_task
    )

关键问题解析 #

1. 原代码wait_for_completion=False无效原因
   原代码在DAG顶层调用API并创建TriggerDagRunOperator，导致Airflow每次解析DAG文件都会执行API调用（增加解析负担）。另外，wait_for_completion=False的作用是让单个触发任务在完成子DAG触发后立即标记成功，Controller DAG会在所有触发任务完成后结束——若你观察到Controller DAG等待子DAG完成，大概率是参数未正确设置（如拼写错误）、Airflow版本过低，或混淆了Controller DAG任务完成与子DAG完成的概念。

2. @task内实例化TriggerDagRunOperator无效原因
   TriggerDagRunOperator是Airflow的调度单元，需由调度器识别并执行，不能在Python任务函数内部直接实例化调用——仅创建对象不会触发任何DAG运行，必须通过Airflow API（如本地客户端、DagRun.create）实现任务内触发。

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