Airflow触发规则咨询：单上游运行其余成功及比例触发需求

场景1：除一个上游任务运行外，其余均成功时触发 #

Airflow内置的触发规则（如all_success、one_success等）没有直接匹配该场景的选项，需要通过自定义逻辑实现：

• 推荐使用PythonOperator编写状态检查逻辑，在任务执行前验证上游任务状态：
  1. 从上下文获取当前DAG的运行时间execution_date
  2. 遍历所有上游任务，查询它们的实例状态
  3. 统计success和running状态的任务数量
  4. 当成功数等于上游总任务数减1，且运行数等于1时，执行当前任务逻辑；否则抛出异常终止任务

示例代码片段：

from airflow.models import TaskInstance
from airflow.operators.python import PythonOperator

def check_upstream_status(**context):
    execution_date = context['execution_date']
    upstream_tasks = context['task'].upstream_list
    total_upstream = len(upstream_tasks)
    
    success_count = 0
    running_count = 0
    
    for task in upstream_tasks:
        ti = TaskInstance(task, execution_date)
        if ti.state == 'success':
            success_count += 1
        elif ti.state == 'running':
            running_count += 1
    
    if success_count == total_upstream - 1 and running_count == 1:
        # 满足触发条件，执行业务逻辑
        pass
    else:
        raise ValueError("上游任务状态不符合触发要求")

# 配置任务
check_trigger_task = PythonOperator(
    task_id='check_upstream_trigger',
    python_callable=check_upstream_status,
    provide_context=True,
    dag=dag
)

场景2：指定比例上游任务成功时触发 #

同样需要自定义逻辑实现比例判断，核心是统计成功任务数占总上游任务数的比例是否达标：

• 步骤与场景1类似，仅调整判断条件为比例阈值（如80%），并处理取整逻辑（如总任务11个时，80%对应9个成功任务）

示例代码片段：

from airflow.models import TaskInstance
from airflow.operators.python import PythonOperator

def check_success_ratio(**context):
    execution_date = context['execution_date']
    upstream_tasks = context['task'].upstream_list
    total_upstream = len(upstream_tasks)
    target_ratio = 0.8  # 设定80%的触发阈值
    
    success_count = 0
    for task in upstream_tasks:
        ti = TaskInstance(task, execution_date)
        if ti.state == 'success':
            success_count += 1
    
    # 计算需要的最少成功任务数（取最接近整数）
    required_success = round(total_upstream * target_ratio)
    
    if success_count >= required_success:
        # 满足比例要求，执行业务逻辑
        pass
    else:
        raise ValueError(f"上游成功任务数未达到{target_ratio*100}%的要求")

# 配置任务
ratio_check_task = PythonOperator(
    task_id='check_success_ratio',
    python_callable=check_success_ratio,
    provide_context=True,
    dag=dag
)

进阶方案：自定义通用触发规则类 #

如果需要在多个任务中复用上述逻辑，可以继承BaseTriggerRule编写自定义触发规则类，直接配置在任务的trigger_rule参数中（Airflow 2.x版本支持）：

from airflow.triggers.base import BaseTriggerRule
from airflow.utils.state import State

class AlmostAllSuccessRunningOne(BaseTriggerRule):
    def __call__(self, task_instance, dag_run, task):
        upstream_task_ids = [t.task_id for t in task.upstream_list]
        upstream_tis = [ti for ti in dag_run.get_task_instances() if ti.task_id in upstream_task_ids]
        
        success_count = sum(1 for ti in upstream_tis if ti.state == State.SUCCESS)
        running_count = sum(1 for ti in upstream_tis if ti.state == State.RUNNING)
        
        return success_count == len(upstream_task_ids) - 1 and running_count == 1

# 使用自定义触发规则
your_business_task = PythonOperator(
    task_id='your_business_task',
    python_callable=your_business_logic,
    trigger_rule=AlmostAllSuccessRunningOne(),
    dag=dag
)

注意事项 #

• 上述逻辑依赖Airflow元数据库的访问权限，需确保任务能正常查询任务实例状态
• 若上游任务为动态生成，需确保upstream_list能正确获取所有上游任务
• 自定义触发规则类仅适用于Airflow 2.x版本，1.x版本需调整为PythonOperator实现

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