如何设置Airflow触发规则：部分上游任务成功且全部完成

可行性结论 #

完全可行，Airflow支持通过两种方式实现该需求：自定义触发规则，或通过辅助任务转换上游任务状态后使用默认触发规则。

实现方法 #

方法一：使用辅助任务（推荐，无需自定义规则） #

核心思路是为task_c添加一个辅助任务，确保无论task_c成功或失败，该辅助任务都会标记为成功。随后让task_d依赖task_a、task_b和这个辅助任务，使用默认的ALL_SUCCESS触发规则即可满足需求。

代码示例：

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="custom_dependency_dag",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None
) as dag:
    # 定义上游任务
    task_a = DummyOperator(task_id="task_a")
    task_b = DummyOperator(task_id="task_b")
    task_c = DummyOperator(task_id="task_c")

    # 辅助任务：task_c完成后（无论成败）执行成功
    task_c_completed = DummyOperator(
        task_id="task_c_completed",
        trigger_rule=TriggerRule.ALL_DONE  # 上游完成即触发，不管状态
    )

    # 下游任务：依赖a、b、c_completed，需全部成功
    task_d = DummyOperator(task_id="task_d")

    # 设置依赖关系
    task_c >> task_c_completed
    [task_a, task_b, task_c_completed] >> task_d

方法二：自定义触发规则 #

如果需要更直接的逻辑控制，可以编写自定义触发规则类，直接检查上游任务的状态是否符合要求。

代码示例：

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

# 自定义触发规则类
class CustomDependencyRule(TriggerRule):
    def is_met(self, task_instance):
        # 获取所有上游任务的状态
        upstream_states = task_instance.get_upstream_states()
        
        # 检查task_a和task_b必须成功
        if upstream_states.get("task_a") != State.SUCCESS or upstream_states.get("task_b") != State.SUCCESS:
            return False
        
        # 检查task_c必须处于终端完成状态（成功/失败/跳过）
        terminal_states = {State.SUCCESS, State.FAILED, State.SKIPPED}
        if upstream_states.get("task_c") not in terminal_states:
            return False
        
        return True

with DAG(
    dag_id="custom_trigger_rule_dag",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None
) as dag:
    task_a = DummyOperator(task_id="task_a")
    task_b = DummyOperator(task_id="task_b")
    task_c = DummyOperator(task_id="task_c")

    # 下游任务使用自定义触发规则
    task_d = DummyOperator(
        task_id="task_d",
        trigger_rule=CustomDependencyRule()
    )

    # 设置依赖关系
    [task_a, task_b, task_c] >> task_d

说明 #

• 方法一的优势是无需编写自定义代码，利用Airflow原生的触发规则即可实现，维护成本更低。
• 方法二更灵活，适合复杂的状态检查场景，但需要维护自定义规则类。

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