TriggerDagRunOperator两次触发同DAG仅成功一次的问题及解决

核心原因是循环中创建的TriggerDagRunOperator任务ID重复：

• Airflow在解析DAG定义时，以task_id作为任务的唯一标识。如果循环生成的多个TriggerDagRunOperator使用了相同的task_id，Airflow会自动覆盖前面的任务实例，最终仅保留最后一个，导致实际执行时只触发一次目标DAG。
• 即便你为每个触发设置了不同的trigger_run_id，只要task_id重复，Airflow就会判定为同一个任务，不会同时存在多个独立的触发任务。

核心是确保循环中每个TriggerDagRunOperator拥有唯一的task_id，同时保证trigger_run_id的唯一性。以下是具体实现示例：

代码示例 #

假设目标DAG的ID为target_dag，循环两次传入不同配置：

from airflow import DAG
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
import uuid
from datetime import datetime

with DAG(
    dag_id="triggering_dag",
    schedule_interval=None,
    start_date=datetime(2024, 1, 1),
    catchup=False
) as dag:
    # 循环生成触发任务，每次传入不同conf
    for idx, conf in enumerate([{"param": "value1"}, {"param": "value2"}]):
        # 生成唯一task_id和trigger_run_id
        unique_suffix = uuid.uuid4().hex[:8]
        TriggerDagRunOperator(
            task_id=f"trigger_target_dag_{idx}_{unique_suffix}",
            trigger_dag_id="target_dag",
            conf=conf,
            reset_dag_run=True,
            execution_date="{{ ts }}",
            trigger_run_id=f"trigger_run_{idx}_{unique_suffix}"
        )

关键注意事项 #

• task_id必须唯一：通过循环索引、UUID片段等拼接方式，确保每个触发任务的task_id不重复。Airflow不允许同一DAG内存在重复的task_id，否则解析阶段会直接覆盖前置任务。
• trigger_run_id保持唯一：结合reset_dag_run=True，唯一的trigger_run_id能保证两次触发生成独立的DAG运行记录，不会互相覆盖。
• 避免循环内复用同一变量名：每次循环都新建独立的Operator实例，确保所有触发任务都被正确添加到DAG中。

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