Airflow任务组遇任务失败立即标记失败，all_done触发规则异常求助

核心原因 #

Airflow默认逻辑是只要Task Group内任一任务失败，就立刻将组状态设为失败，不会等待其余任务执行完毕。这会导致依赖该组、使用all_done触发规则的下游任务提前启动，和你预期的“等组内所有任务结束再处理”完全不符。

可行解决方案 #

1. 自定义组内收尾任务控制状态感知 #

通过在Task Group内添加一个收尾检查任务，手动确认所有任务完成后再触发下游，替代依赖Task Group本身的状态：

• 给组内每个任务添加回调，将执行状态写入XCom
• 用一个Python任务拉取组内所有任务的状态，只有当所有任务都完成（无论成功失败）才允许下游执行
• 下游任务直接依赖这个收尾任务，而非Task Group

示例代码片段：

from airflow.models import TaskInstance
from airflow.operators.python import PythonOperator
from airflow.utils.task_group import TaskGroup
from datetime import timedelta

def check_all_tasks_completed(ti, group_id):
    # 获取组内所有任务实例
    task_instances = TaskInstance.find(
        dag_id=ti.dag_id,
        execution_date=ti.execution_date,
        task_group_id=group_id
    )
    # 检查所有任务是否已脱离运行/排队状态
    all_finished = all(
        ti.state not in ['running', 'queued'] 
        for ti in task_instances
    )
    if not all_finished:
        # 未全部完成则抛出异常触发重试
        raise Exception("Group tasks not all completed yet")
    # 返回汇总状态供下游参考
    return {ti.task_id: ti.state for ti in task_instances}

with TaskGroup("my_task_group") as tg:
    task1 = PythonOperator(
        task_id="task1",
        python_callable=lambda: print("Task1 running"),
        on_success_callback=lambda ti: ti.xcom_push(key="status", value=ti.state),
        on_failure_callback=lambda ti: ti.xcom_push(key="status", value=ti.state)
    )
    task2 = PythonOperator(
        task_id="task2",
        python_callable=lambda: 1/0,  # 模拟任务失败
        on_success_callback=lambda ti: ti.xcom_push(key="status", value=ti.state),
        on_failure_callback=lambda ti: ti.xcom_push(key="status", value=ti.state)
    )
    # 收尾检查任务，设置重试间隔直到所有任务完成
    wait_all_done = PythonOperator(
        task_id="wait_all_done",
        python_callable=check_all_tasks_completed,
        op_kwargs={"group_id": tg.group_id},
        retries=60,
        retry_delay=timedelta(minutes=1),
        trigger_rule="all_done"
    )
    # 组内任务并行执行，最终都指向检查任务
    [task1, task2] >> wait_all_done

# 下游任务依赖收尾检查任务，而非Task Group
final_task = PythonOperator(
    task_id="final_task",
    python_callable=lambda: print("All group tasks finished, running final task"),
    trigger_rule="all_done"
)
wait_all_done >> final_task

2. 用ExternalTaskSensor监听组内所有任务 #

如果不想修改组内任务逻辑，可以在下游前添加传感器，监听组内每个任务的状态，仅当所有任务都进入完成状态（成功/失败/跳过）时才触发下游：

from airflow.sensors.external_task import ExternalTaskSensor

group_task_sensor = ExternalTaskSensor(
    task_id="wait_group_tasks",
    external_dag_id=dag.dag_id,
    external_task_ids=["task1", "task2"],  # 列出组内所有任务ID
    allowed_states=["success", "failed", "skipped"],
    failed_states=[],
    poke_interval=60,
    mode="reschedule"
)
group_task_sensor >> final_task

3. 修改Airflow核心代码（不推荐） #

Task Group的状态计算逻辑在airflow/utils/task_group.py的get_state方法中，如果你有部署权限，可以修改该方法让组状态仅在所有任务完成后更新。但这种方式会影响全局Task Group行为，且Airflow升级时会丢失修改，不建议生产环境使用。

关键注意事项 #

• 不要依赖Task Group的状态触发下游，直接依赖收尾任务或传感器更可靠
• 收尾任务的重试次数和间隔要根据组内任务的最长执行时间合理设置
• 确保组内所有任务都正确上报状态到XCom，避免遗漏导致收尾任务无限重试

内容的提问来源于stack exchange，提问作者Márcio Marchiori