Airflow如何实现仅当任务D成功时触发任务G且失败即终止DAG？

针对你描述的Airflow场景，以下是具体的实现方案：

核心思路 #

要实现仅当任务D成功走完分支（D→E→F全成功）才执行G，同时保证任何任务失败则整个DAG标记为失败，需要从依赖关系和触发规则两方面调整：

1. 让G的触发严格依赖F的成功状态，而非上游分支节点B；
2. 利用Airflow默认的任务失败传导机制，确保单个任务失败会直接标记DAG为失败。

具体代码实现 #

from airflow import DAG
from airflow.operators.python import PythonOperator, BranchPythonOperator
from airflow.utils.trigger_rule import TriggerRule
from datetime import datetime

# 分支逻辑函数：根据实际业务条件返回要执行的任务ID
def branch_decision():
    # 示例条件：可替换为你的业务判断逻辑
    should_run_c = False
    return "task_C" if should_run_c else "task_D"

# 各任务的执行函数
def run_task_a():
    print("执行任务A")

def run_task_b():
    print("执行任务B")

def run_task_c():
    print("执行任务C")

def run_task_d():
    print("执行任务D")

def run_task_e():
    print("执行任务E")

def run_task_f():
    print("执行任务F")

def run_task_g():
    print("执行任务G")

default_args = {
    "owner": "airflow",
    "start_date": datetime(2024, 1, 1),
    "retries": 0  # 根据需求设置重试次数，重试失败后任务会标记为失败
}

with DAG(
    dag_id="conditional_g_execution",
    default_args=default_args,
    schedule_interval=None,
    catchup=False
) as dag:
    task_A = PythonOperator(
        task_id="task_A",
        python_callable=run_task_a
    )

    task_B = BranchPythonOperator(
        task_id="task_B",
        python_callable=branch_decision
    )

    task_C = PythonOperator(
        task_id="task_C",
        python_callable=run_task_c
    )

    task_D = PythonOperator(
        task_id="task_D",
        python_callable=run_task_d
    )

    task_E = PythonOperator(
        task_id="task_E",
        python_callable=run_task_e
    )

    task_F = PythonOperator(
        task_id="task_F",
        python_callable=run_task_f
    )

    task_G = PythonOperator(
        task_id="task_G",
        python_callable=run_task_g,
        # 触发规则使用默认的ALL_SUCCESS，可省略不写，明确声明更清晰
        trigger_rule=TriggerRule.ALL_SUCCESS
    )

    # 构建依赖关系
    task_A >> task_B
    task_B >> [task_C, task_D]
    task_D >> task_E >> task_F >> task_G

关键细节说明 #

1. 分支逻辑处理：
   用BranchPythonOperator实现任务B的分支判断，每次只会触发C或D中的一个任务，另一个会被标记为skipped（跳过），不会影响DAG状态。

2. G的触发控制：
   G仅依赖任务F，且使用ALL_SUCCESS触发规则：

   • 当流程走D分支且D、E、F全成功时，F状态为success，G会正常执行；
   • 当流程走C分支时，D、E、F都会被标记为skipped，G因依赖的F未成功，会保持skipped状态，不会执行。

3. DAG失败传导：
   Airflow默认行为是：只要DAG中有任意任务标记为failed，整个DAG的状态就会变为Failed。不需要额外配置，只要不修改任务的触发规则为允许失败的类型（如all_done、none_failed）即可。

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