TriggerDagRunOperator失败后如何实现DAG持续运行？

方案1：给TriggerDagRunOperator添加重试机制 #

直接给Operator配置重试参数，让它在失败后自动重试，直到成功触发下一次DAG运行，避免单次失败中断循环：

from airflow.operators.trigger_dagrun import TriggerDagRunOperator
from datetime import timedelta

trigger_next_run = TriggerDagRunOperator(
    task_id="trigger_next_dag",
    trigger_dag_id="your_target_dag_id",
    retries=5,  # 根据实际情况调整重试次数
    retry_delay=timedelta(minutes=3),  # 设置重试间隔
    dag=dag
)

该方案适用于TriggerDagRunOperator仅偶尔失败的场景，重试后大概率能恢复触发逻辑。

方案2：用PythonOperator自定义触发逻辑并捕获异常 #

如果TriggerDagRunOperator的失败难以通过重试解决，或需要更灵活的控制，可以用PythonOperator替代，手动实现触发逻辑并捕获所有异常，确保任务始终标记为成功，维持DAG的循环运行：

from airflow.operators.python import PythonOperator
from airflow.models import DagRun
from airflow.utils.state import State
from airflow.utils.dates import days_ago

def trigger_self_dag(**context):
    try:
        target_dag_id = context["dag"].dag_id
        # 创建新的DAG运行实例
        DagRun.create(
            dag_id=target_dag_id,
            execution_date=days_ago(0),
            state=State.RUNNING
        )
        context["ti"].log.info("成功触发下一次DAG运行")
    except Exception as e:
        # 捕获所有异常，仅记录日志不抛出，确保任务状态为成功
        context["ti"].log.error(f"触发DAG失败: {str(e)}")
        # 可选：添加告警逻辑（如邮件、Slack通知），但不影响任务状态

trigger_task = PythonOperator(
    task_id="trigger_self",
    python_callable=trigger_self_dag,
    provide_context=True,
    dag=dag
)

这种方式完全绕过TriggerDagRunOperator的失败判定，只要任务本身不失败，DAG就能正常结束并进入下一次循环。

方案3：设置DAG全局失败回调兜底 #

如果整个DAG因TriggerDagRunOperator失败而标记为失败，可以给DAG配置on_failure_callback，在DAG失败时自动触发自身，作为最后一层兜底：

from airflow.models import DAG, DagRun
from airflow.utils.state import State
from airflow.utils.dates import days_ago

def trigger_on_dag_failure(context):
    target_dag_id = context["dag"].dag_id
    DagRun.create(
        dag_id=target_dag_id,
        execution_date=days_ago(0),
        state=State.RUNNING
    )
    context["dag"].log.info("DAG失败，已自动触发下一次运行")

default_args = {
    'owner': 'airflow',
    'start_date': days_ago(1),
    'on_failure_callback': trigger_on_dag_failure
}

dag = DAG(
    'your_dag_id',
    default_args=default_args,
    schedule_interval='once'
)

该方案能确保无论DAG因何失败都能启动下一次运行，但需注意排查重复失败的根源，避免陷入无效的失败循环。

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