Airflow单DAG内实现task1失败则全DAG终止的配置方案咨询

任务依赖结构 #

task1 (must succeed)
            |
            v
   +-------------------+
   |                   |
task2-A             task2-B
   |                   |
   v                   v
task3-A             task3-B
       \             /
        \           /
         v         v
    +-------------------+
    |                   |
  task4-A           task4-B
    |                   |
    +-------------------+
    |                   |
  task5-A            task5-B
         \         /
          \       /
           v     v
            task6
              |
              v
           task7
              |
              v
           task8

需求与现存问题 #

核心需求 #

• task1失败则整个DAG直接终止，所有后续任务禁止执行
• task1成功后，后续任务无需等待自身前置任务的成功状态，只要资源允许即可启动
• 同一时间最多仅允许2个任务运行

现存问题 #

task1失败时，task2-A/B、3-A/B因直接依赖task1而失败，但task4-A/B因设置了ALL_DONE触发规则仍会启动；尝试分支逻辑与多种触发规则组合均无效，目前只能通过两个独立DAG实现。

单个DAG内的解决方案 #

核心思路 #

给所有后续任务绑定task1成功依赖作为全局开关，同时针对各层级任务调整触发规则，既保证task1失败时全局终止，又满足task1成功后任务无需依赖前置状态即可运行的要求，最后通过DAG参数控制并发数。

具体配置步骤 #

1. 全局开关绑定：所有后续任务（task2-A/B至task8）都添加对task1的成功依赖，确保只有task1成功时，这些任务才有执行可能。
2. 触发规则适配：
   • task2-A/B：保持默认ALL_SUCCESS触发规则，仅在task1成功时启动
   • task3-A/B至task8：设置触发规则为ALL_DONE，同时绑定对应前置任务与task1——只要task1成功，不管直接前置任务成功/失败，当前任务均可启动
3. 并发控制：在DAG定义中设置concurrency=2，限制同一时间运行的任务数量。

代码示例 #

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.utils.trigger_rule import TriggerRule
from datetime import datetime

with DAG(
    dag_id="unified_task_dag",
    start_date=datetime(2024, 1, 1),
    concurrency=2,  # 限制同时运行的任务数
    catchup=False,
    default_args={"retries": 0}
) as dag:
    # 全局入口任务
    task1 = DummyOperator(task_id="task1")

    # 第一层级任务：仅依赖task1成功
    task2_a = DummyOperator(task_id="task2-A")
    task2_b = DummyOperator(task_id="task2-B")
    task1 >> [task2_a, task2_b]

    # 第二层级任务：依赖task1+对应task2，前置完成即可启动
    task3_a = DummyOperator(task_id="task3-A", trigger_rule=TriggerRule.ALL_DONE)
    task3_b = DummyOperator(task_id="task3-B", trigger_rule=TriggerRule.ALL_DONE)
    [task1, task2_a] >> task3_a
    [task1, task2_b] >> task3_b

    # 第三层级任务
    task4_a = DummyOperator(task_id="task4-A", trigger_rule=TriggerRule.ALL_DONE)
    task4_b = DummyOperator(task_id="task4-B", trigger_rule=TriggerRule.ALL_DONE)
    [task1, task3_a] >> task4_a
    [task1, task3_b] >> task4_b

    # 第四层级任务
    task5_a = DummyOperator(task_id="task5-A", trigger_rule=TriggerRule.ALL_DONE)
    task5_b = DummyOperator(task_id="task5-B", trigger_rule=TriggerRule.ALL_DONE)
    [task1, task4_a] >> task5_a
    [task1, task4_b] >> task5_b

    # 最终串联任务
    task6 = DummyOperator(task_id="task6", trigger_rule=TriggerRule.ALL_DONE)
    [task1, task5_a, task5_b] >> task6

    task7 = DummyOperator(task_id="task7", trigger_rule=TriggerRule.ALL_DONE)
    [task1, task6] >> task7

    task8 = DummyOperator(task_id="task8", trigger_rule=TriggerRule.ALL_DONE)
    [task1, task7] >> task8

方案验证 #

• task1失败时：所有后续任务因未满足task1的成功依赖，均处于upstream_failed状态，不会启动
• task1成功时：各层级任务只要其直接前置任务完成（无论成功/失败），即可在资源允许的情况下启动，符合需求
• 并发数控制：concurrency=2会自动限制DAG内同时运行的任务数量，满足资源限制要求

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