如何实现：三个任务中两任务失败时触发DAG失败的判定逻辑

要满足你的需求，需从任务依赖触发规则和状态判定逻辑两部分入手，以下是基于Airflow框架的具体实现步骤：

1. 核心配置要点 #

• 确保state任务在所有task_x完成后执行：使用trigger_rule='all_done'（默认规则为all_success，仅当所有上游成功才执行，不符合需求）。
• 在state任务中统计失败任务数：通过Airflow的TaskInstance API获取上游任务状态，判断失败数量是否≥2，若是则主动让state任务失败，从而终止整个DAG。

2. 完整代码示例 #

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.utils.state import State
from datetime import datetime

def check_task_failure_threshold(**context):
    # 获取当前任务的TaskInstance对象
    ti = context['ti']
    # 定义需要检查的上游任务ID列表
    target_tasks = ['task_1', 'task_2', 'task_3']
    failed_task_count = 0

    # 遍历统计失败任务数量
    for task_id in target_tasks:
        upstream_ti = ti.get_task_instance(task_id=task_id)
        if upstream_ti.state == State.FAILED:
            failed_task_count += 1

    # 判定失败数量是否达标，达标则抛出异常终止DAG
    if failed_task_count >= 2:
        raise RuntimeError(f"已检测到{failed_task_count}个任务失败，达到阈值（2个），DAG终止")

# 定义DAG基础配置
with DAG(
    dag_id="multi_task_failure_check",
    schedule_interval=None,
    start_date=datetime(2024, 1, 1),
    catchup=False,
    default_args={"owner": "airflow"}
) as dag:
    # 模拟三个业务任务，可替换为实际任务逻辑
    task_1 = BashOperator(
        task_id="task_1",
        bash_command="exit 0"  # 修改为exit 1可模拟任务失败
    )

    task_2 = BashOperator(
        task_id="task_2",
        bash_command="exit 0"
    )

    task_3 = BashOperator(
        task_id="task_3",
        bash_command="exit 0"
    )

    # 状态判定任务
    state_task = PythonOperator(
        task_id="state",
        python_callable=check_task_failure_threshold,
        trigger_rule="all_done",  # 关键：所有上游任务完成后执行，无论成功/失败
        provide_context=True  # Airflow 2.x可省略，默认传递context
    )

# 设置任务依赖
task_1 >> state_task
task_2 >> state_task
task_3 >> state_task

3. 关键逻辑说明 #

• trigger_rule='all_done'：强制state任务在task_1/task_2/task_3全部执行完成后启动，不管这些任务是成功、失败还是跳过。
• 状态获取与判定：通过ti.get_task_instance()从Airflow元数据库中读取上游任务的实际执行状态，统计失败数量后，若达到阈值则抛出异常，让state任务标记为失败，进而导致整个DAG运行失败。
• 测试验证：修改任意两个BashOperator的bash_command为exit 1，触发任务失败，即可验证state任务是否会终止DAG。

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