如何在Airflow中配置自定义触发规则？特定任务触发场景求助

针对你提出的「仅当Task1失败，且Task2、Task3执行成功时触发Task4」的场景，Airflow自带的触发规则确实没有直接匹配的选项，不过可以通过以下两种方案实现需求：

方案一：新增前置判断任务（兼容所有Airflow版本） #

通过新增一个用于状态校验的任务，先判断上游三个任务的状态是否符合要求，再决定是否触发Task4：

1. 添加一个PythonOperator或ShortCircuitOperator任务（示例用PythonOperator），依赖Task1、Task2、Task3
2. 在该任务的执行逻辑中，检查三个上游任务的状态：Task1需为FAILED，Task2和Task3需为SUCCESS
3. 满足条件时任务执行成功，Task4依赖这个校验任务（使用默认的all_success触发规则即可）；不满足条件时让校验任务失败，Task4就不会被触发

代码示例 #

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.dummy import DummyOperator
from airflow.utils.state import State
from datetime import datetime

def check_upstream_states(**context):
    # 获取当前DAG运行中所有任务实例的状态
    task_instances = context['ti'].get_dagrun().get_task_instances()
    task_state_map = {ti.task_id: ti.state for ti in task_instances}
    
    # 校验目标状态条件
    if (task_state_map.get('Task1') == State.FAILED and
        task_state_map.get('Task2') == State.SUCCESS and
        task_state_map.get('Task3') == State.SUCCESS):
        return
    # 不满足条件时抛出异常，标记任务失败
    raise ValueError("上游任务状态不符合触发Task4的要求")

with DAG(
    dag_id="custom_condition_trigger",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
    catchup=False
) as dag:
    Task1 = DummyOperator(task_id="Task1")
    Task2 = DummyOperator(task_id="Task2")
    Task3 = DummyOperator(task_id="Task3")
    
    check_condition = PythonOperator(
        task_id="check_upstream_condition",
        python_callable=check_upstream_states,
        provide_context=True
    )
    
    Task4 = DummyOperator(task_id="Task4")
    
    [Task1, Task2, Task3] >> check_condition >> Task4

方案二：自定义TriggerRule（仅支持Airflow 2.x+） #

Airflow 2.x及以上版本允许自定义触发规则，通过继承BaseTriggerRule类实现专属的触发逻辑：

代码示例 #

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.triggers.base import BaseTriggerRule
from airflow.utils.state import State
from datetime import datetime

class Task1FailOthersSuccessTrigger(BaseTriggerRule):
    def should_trigger(self, task_states):
        # task_states是上游任务ID与对应状态的字典
        return (task_states.get("Task1") == State.FAILED and
                task_states.get("Task2") == State.SUCCESS and
                task_states.get("Task3") == State.SUCCESS)

with DAG(
    dag_id="custom_trigger_rule_dag",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
    catchup=False
) as dag:
    Task1 = DummyOperator(task_id="Task1")
    Task2 = DummyOperator(task_id="Task2")
    Task3 = DummyOperator(task_id="Task3")
    
    Task4 = DummyOperator(
        task_id="Task4",
        trigger_rule=Task1FailOthersSuccessTrigger()
    )
    
    [Task1, Task2, Task3] >> Task4

注意事项 #

• 方案一兼容性更强，适用于所有Airflow版本；方案二需要Airflow 2.x及以上版本支持
• 测试时可通过手动标记上游任务的状态（在Airflow UI中操作），验证触发逻辑是否符合预期

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