如何在Airflow中实现one_done触发规则？

针对你提到的依赖场景（A >> [B,C,D,E] >> F，需F在B/C/D/E任一完成（成功或失败均可）时立即触发），Airflow确实没有内置的one_done触发规则，但可以通过以下两种方式实现：

方法一：自定义触发规则（最直接方案） #

Airflow 2.x支持自定义触发规则，只需继承BaseTriggerRule并实现判断逻辑即可：

from airflow.utils.trigger_rule import BaseTriggerRule
from airflow.models.taskinstance import TaskInstance
from airflow.utils.state import State

class OneDoneTriggerRule(BaseTriggerRule):
    def should_trigger(self, task_instances: list[TaskInstance]) -> bool:
        # 只要有一个上游任务处于已完成状态（成功/失败都算），就触发当前任务
        completed_states = {State.SUCCESS, State.FAILED}
        return any(ti.state in completed_states for ti in task_instances)

然后在定义任务F时，指定这个自定义触发规则：

from airflow.operators.dummy import DummyOperator

# 定义任务
A = DummyOperator(task_id="task_A")
B = DummyOperator(task_id="task_B")
C = DummyOperator(task_id="task_C")
D = DummyOperator(task_id="task_D")
E = DummyOperator(task_id="task_E")
F = DummyOperator(
    task_id="task_F",
    trigger_rule=OneDoneTriggerRule()  # 使用自定义触发规则
)

# 设置依赖
A >> [B, C, D, E] >> F

方法二：借助分支任务+状态检查（无需自定义类） #

如果不想编写自定义触发规则类，可以给每个上游任务（B/C/D/E）都添加分支检查逻辑，当任务完成时直接触发F，同时给F设置max_active_runs=1避免重复执行：

from airflow.operators.dummy import DummyOperator
from airflow.operators.python import BranchPythonOperator
from airflow.utils.state import State

def check_task_completed(**context):
    task_id = context["params"]["task_id"]
    ti = context["ti"].dag_run.get_task_instance(task_id)
    if ti.state in {State.SUCCESS, State.FAILED}:
        return "task_F"
    return "dummy_wait"

# 定义任务
A = DummyOperator(task_id="task_A")
B = DummyOperator(task_id="task_B")
C = DummyOperator(task_id="task_C")
D = DummyOperator(task_id="task_D")
E = DummyOperator(task_id="task_E")
F = DummyOperator(task_id="task_F", max_active_runs=1)
dummy_wait = DummyOperator(task_id="dummy_wait")

# 为每个上游任务创建分支检查任务
branch_B = BranchPythonOperator(
    task_id="branch_check_B",
    python_callable=check_task_completed,
    params={"task_id": "task_B"},
    provide_context=True
)
branch_C = BranchPythonOperator(
    task_id="branch_check_C",
    python_callable=check_task_completed,
    params={"task_id": "task_C"},
    provide_context=True
)
branch_D = BranchPythonOperator(
    task_id="branch_check_D",
    python_callable=check_task_completed,
    params={"task_id": "task_D"},
    provide_context=True
)
branch_E = BranchPythonOperator(
    task_id="branch_check_E",
    python_callable=check_task_completed,
    params={"task_id": "task_E"},
    provide_context=True
)

# 设置依赖
A >> [B, C, D, E]
B >> branch_B >> [F, dummy_wait]
C >> branch_C >> [F, dummy_wait]
D >> branch_D >> [F, dummy_wait]
E >> branch_E >> [F, dummy_wait]

注意事项 #

• 自定义触发规则方案更简洁，推荐在Airflow 2.x及以上版本使用；
• 方法二中设置max_active_runs=1是为了防止多个上游任务完成时重复触发F；
• 两种方案都能实现“任一上游完成即触发F”的需求，无论上游任务是成功还是失败状态。

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