Airflow：检查首个上游Operator是否跳过并触发流水线失败

Airflow 自带的 trigger_rule 确实没有直接匹配「仅前一个上游被跳过就触发当前任务失败」的规则——trigger_rule 的作用是控制任务是否执行，而非在执行时根据上游状态触发失败逻辑。你可以通过以下两种方式实现需求：

方法一：用 PythonOperator 快速实现检查逻辑 #

直接编写检查函数，获取前一个上游任务的状态，若为SKIPPED则抛出异常让当前任务失败，进而带动整个流水线失败。

代码示例 #

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.models import TaskInstance, State
from airflow.utils.trigger_rule import TriggerRule
from datetime import datetime

def check_previous_task_status(**context):
    # 获取当前任务的首个上游任务（链式场景下上游仅一个）
    upstream_task = context['task'].upstream_list[0]
    # 获取当前DAG运行的执行时间
    execution_date = context['execution_date']
    # 实例化上游任务的TaskInstance并从数据库刷新状态
    ti = TaskInstance(task=upstream_task, execution_date=execution_date)
    ti.refresh_from_db()
    
    # 检查上游状态是否为SKIPPED，是则抛出异常触发失败
    if ti.state == State.SKIPPED:
        raise ValueError(f"上游任务 {upstream_task.task_id} 被跳过，触发流水线失败")

# 定义DAG
with DAG(
    dag_id='chain_with_skip_check',
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
) as dag:
    task1 = PythonOperator(
        task_id='task1',
        python_callable=lambda: print("Task1 executed")
    )
    
    # 检查任务：设置trigger_rule为ALL_DONE，确保上游无论状态如何都会执行检查
    check_skip = PythonOperator(
        task_id='check_task1_skip',
        python_callable=check_previous_task_status,
        trigger_rule=TriggerRule.ALL_DONE,
        provide_context=True
    )
    
    task2 = PythonOperator(
        task_id='task2',
        python_callable=lambda: print("Task2 executed")
    )
    
    # 链式任务
    task1 >> check_skip >> task2

方法二：自定义可复用 Operator #

如果需要在多个任务链中复用该逻辑，可以自定义一个Operator，封装检查逻辑：

代码示例 #

from airflow.models.baseoperator import BaseOperator
from airflow.models import TaskInstance, State
from airflow.utils.context import Context
from airflow.utils.trigger_rule import TriggerRule

class CheckPreviousSkippedOperator(BaseOperator):
    def execute(self, context: Context):
        # 确保上游仅一个任务（链式场景）
        if len(self.upstream_list) != 1:
            raise ValueError("该Operator仅支持单个上游任务的链式场景")
        
        upstream_task = self.upstream_list[0]
        execution_date = context['execution_date']
        ti = TaskInstance(task=upstream_task, execution_date=execution_date)
        ti.refresh_from_db()
        
        if ti.state == State.SKIPPED:
            self.log.error(f"上游任务 {upstream_task.task_id} 状态为SKIPPED，触发失败")
            raise ValueError(f"上游任务 {upstream_task.task_id} 被跳过")

# 在DAG中使用
with DAG(
    dag_id='custom_operator_skip_check',
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
) as dag:
    task1 = PythonOperator(task_id='task1', python_callable=lambda: print("Task1 executed"))
    
    check_skip = CheckPreviousSkippedOperator(
        task_id='check_task1_skip',
        trigger_rule=TriggerRule.ALL_DONE
    )
    
    task2 = PythonOperator(task_id='task2', python_callable=lambda: print("Task2 executed"))
    
    task1 >> check_skip >> task2

关键注意事项 #

• 必须给检查任务设置 trigger_rule=TriggerRule.ALL_DONE：默认的all_success会导致上游被跳过时，检查任务直接被跳过，无法执行检查逻辑。ALL_DONE确保上游任务无论成功、失败还是跳过，检查任务都会执行。
• 链式场景下上游仅一个任务：代码中默认取upstream_list[0]，如果你的场景存在多个上游，需要调整逻辑指定要检查的目标任务。
• 失败传导：检查任务失败后，下游任务会因依赖关系（默认all_success规则）而失败，最终整个流水线状态变为failed。

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