当前DAG运行失败时，如何立即触发新实例且无需调度？

问题解答 #

是的，存在方法在DAG运行失败时触发新的运行实例，并且可以不依赖调度机制立即触发，具体实现方式如下：

1. 利用DAG全局失败回调（最直接方式） #

在DAG的默认参数中配置on_failure_callback，这个回调会在整个DAG运行失败时触发（无论哪个任务失败导致DAG进入失败状态）。在回调函数里，通过Airflow的Python API直接创建新的DAG运行实例，完全不依赖调度器的定时触发逻辑。

示例代码：

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

def trigger_fallback_run(context):
    # 获取当前DAG对象
    target_dag = context["dag"]
    # 创建新的DAG运行实例
    DagRun.create(
        dag_id=target_dag.dag_id,
        execution_date=days_ago(0),  # 使用当前时间作为执行日期
        state=State.RUNNING,
        conf=context["dag_run"].conf  # 继承原运行的配置参数
    )

# 定义DAG时绑定回调
default_args = {
    "owner": "airflow",
    "on_failure_callback": trigger_fallback_run,
    "start_date": days_ago(1)
}

with DAG(
    dag_id="auto_retry_dag",
    default_args=default_args,
    schedule_interval="@daily"
) as dag:
    # 此处添加你的任务定义
    pass

2. 任务级失败回调（单个任务失败即触发） #

如果需要在任意单个任务失败时就触发新DAG运行（无需等待整个DAG进入失败状态），可以给每个任务单独配置on_failure_callback，逻辑和上述一致，只是回调绑定在Task级别而非DAG全局。

关键注意事项 #

• 避免无限循环：必须在回调中添加限制逻辑，比如记录触发次数、检查当前是否已有RUNNING状态的DAG实例，防止因持续失败导致无限触发。
• 权限控制：确保执行回调的Airflow worker账号拥有创建DAG运行的权限。
• Executor兼容性：若使用CeleryExecutor等分布式执行器，需确保worker节点能正常访问Airflow元数据库，避免API调用失败。

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