如何用TriggerDagRunOperator多次触发带不同配置的同一DAG？

你遇到的核心问题是：两次触发都使用了相同的execution_date（父DAG的logical_date），Airflow会将其识别为同一个DAG实例，即使传入的conf不同，也只会重置并重新运行该实例，而不会创建新实例。

要实现不同配置触发同一DAG的独立实例，关键是让每次触发的execution_date唯一，同时让ExternalTaskSensor对应监听该唯一的execution_date。以下是两种可行方案：

方案1：利用XCom传递触发后的实际execution_date（推荐） #

TriggerDagRunOperator执行后会返回触发的DAG Run的元数据（包括execution_date），我们可以通过XCom将其传递给ExternalTaskSensor，让Sensor精准监听对应的实例。

修改后的代码示例：

from airflow.utils.dates import days_ago
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.utils.task_group import TaskGroup
from airflow.decorators import dag

@dag(
    start_date=days_ago(1),
    schedule_interval=None,
    catchup=False
)
def parent_dag():
    @task_group(group_id='refresh_pre-prod')
    def refresh_pre_prod():
        prod_to_pre_prod = TriggerDagRunOperator (
            task_id='prod_to_pre_prod',
            trigger_dag_id="util_clone_bq_env",
            # 不指定execution_date，让Airflow自动生成触发时的时间作为logical_date
            conf={
                    "src_project_id":"production",
                    "trg_project_id":"pre-production"
                 }, 
            reset_dag_run=False,  # 不需要重置，因为是新实例
            wait_for_completion=False  # 不需要等待完成，交给Sensor监听
        )

        prod_to_pre_prod_sensor = ExternalTaskSensor(
            task_id='prod_to_pre_prod_sensor',
            external_dag_id='util_clone_bq_env',
            external_task_id='notify_completion',
            # 从XCom获取Trigger任务返回的execution_date
            execution_date_fn=lambda context: context['ti'].xcom_pull(task_ids='prod_to_pre_prod')['execution_date'],
            allowed_states=["success"],
            failed_states=["failed", "skipped", "upstream_failed"]
        )

        prod_to_pre_prod >> prod_to_pre_prod_sensor

    @task_group(group_id='refresh_demo')
    def refresh_demo():
        prod_to_demo = TriggerDagRunOperator(
            task_id='prod_to_demo',
            trigger_dag_id="util_clone_bq_env",
            conf={
                    "src_project_id":"production",
                    "trg_project_id":"demo1"
                 },
            reset_dag_run=False,
            wait_for_completion=False
        )

        prod_to_demo_sensor = ExternalTaskSensor(
            task_id='prod_to_demo_sensor',
            external_dag_id='util_clone_bq_env',
            external_task_id='notify_completion',
            execution_date_fn=lambda context: context['ti'].xcom_pull(task_ids='prod_to_demo')['execution_date'],
            allowed_states=["success"],
            failed_states=["failed", "skipped", "upstream_failed"]
        )

        prod_to_demo >> prod_to_demo_sensor

    refresh_pre_prod()
    refresh_demo()

parent_dag()

方案2：手动生成唯一的execution_date #

通过给父DAG的logical_date添加微小时间偏移，确保每次触发的execution_date唯一，同时让Sensor使用相同的偏移逻辑监听对应实例。

修改后的代码示例：

from airflow.utils.dates import days_ago
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.utils.task_group import TaskGroup
from airflow.decorators import dag
from airflow import macros

@dag(
    start_date=days_ago(1),
    schedule_interval=None,
    catchup=False
)
def parent_dag():
    @task_group(group_id='refresh_pre-prod')
    def refresh_pre_prod():
        prod_to_pre_prod = TriggerDagRunOperator (
            task_id='prod_to_pre_prod',
            trigger_dag_id="util_clone_bq_env",
            # 添加1秒偏移，生成唯一execution_date
            execution_date='{{ dag_run.logical_date + macros.timedelta(seconds=1) }}',
            conf={
                    "src_project_id":"production",
                    "trg_project_id":"pre-production"
                 }, 
            reset_dag_run=True
        )

        prod_to_pre_prod_sensor = ExternalTaskSensor(
            task_id='prod_to_pre_prod_sensor',
            external_dag_id='util_clone_bq_env',
            external_task_id='notify_completion',
            # 使用相同的偏移逻辑匹配execution_date
            execution_date='{{ dag_run.logical_date + macros.timedelta(seconds=1) }}',
            allowed_states=["success"],
            failed_states=["failed", "skipped", "upstream_failed"]
        )

        prod_to_pre_prod >> prod_to_pre_prod_sensor

    @task_group(group_id='refresh_demo')
    def refresh_demo():
        prod_to_demo = TriggerDagRunOperator(
            task_id='prod_to_demo',
            trigger_dag_id="util_clone_bq_env",
            # 添加2秒偏移，确保与pre-prod的execution_date不重复
            execution_date='{{ dag_run.logical_date + macros.timedelta(seconds=2) }}',
            conf={
                    "src_project_id":"production",
                    "trg_project_id":"demo1"
                 },
            reset_dag_run=True
        )

        prod_to_demo_sensor = ExternalTaskSensor(
            task_id='prod_to_demo_sensor',
            external_dag_id='util_clone_bq_env',
            external_task_id='notify_completion',
            execution_date='{{ dag_run.logical_date + macros.timedelta(seconds=2) }}',
            allowed_states=["success"],
            failed_states=["failed", "skipped", "upstream_failed"]
        )

        prod_to_demo >> prod_to_demo_sensor

    refresh_pre_prod()
    refresh_demo()

parent_dag()

• 方案1更灵活，无需手动维护时间偏移，适合动态触发场景；
• 方案2需要确保偏移量足够区分不同触发任务，避免execution_date冲突；
• 若目标DAG（util_clone_bq_env）依赖execution_date做业务逻辑，需确认自定义的execution_date不影响其功能；
• 确保目标DAG的catchup设置为False，避免自动补历史任务。

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