Airflow v1 和 v2 之间“跳过”状态传播的行为变化？答案

【问题标题】：Behavior change in 'skipped' status propagation between Airflow v1 and v2?Airflow v1 和 v2 之间“跳过”状态传播的行为变化？
【发布时间】：2022-01-05 15:59:36
【问题描述】：

我注意到，在 Airflow 1 中，当任务失败时，以下任务具有 upstream_failed 状态，但在 ALL_SUCCESS 和 ONE_SUCCESS 触发规则的情况下，紧随其后的任务具有 skipped 状态。

这适用于具有NONE_SKIPPED 触发规则的处理程序任务，只有在没有跳过其前面的任务时才应该执行的任务（因此它们可以具有succces、failed 或upstream_failed 状态），当其前面的任何任务为skipped 时，它被跳过。这很好，因为upstream_failed 状态的任务后面跟着skipped 任务。

气流1：

在 Airflow 2 中情况有所不同，因为相同的触发规则配置工作方式不同（我不明白这种变化）

处理程序在每种情况下都会被执行，因为failed 状态总是跟在upstream_failed 状态之后，而upstream_failed 状态永远不会跟在skipped 状态之后。

气流2：

这两个版本之间发生重大变化的原因是什么？
如何在 Airflow 2 中获得与 Airflow 1 中已有的相同的行为？

这是我正在玩的测试代码：

import datetime
from datetime import timedelta

from airflow.models import DAG
from airflow.utils.trigger_rule import TriggerRule
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator


def fail_this_task():
    raise Exception("I am angry at Airflow, so I deliberately fail.")

dag_args = {
    'start_date': datetime.datetime.combine(datetime.datetime.today() - datetime.timedelta(1), datetime.datetime.min.time()),
    'email_on_failure': False,
    'email_on_retry': False,
    'catchup' : False,
    'retries': 0,
}

with DAG(
    default_args        = dag_args,
    dag_id              = 'my_annoying_dag',
    schedule_interval   = None
) as dag:

    FailDeliberately = PythonOperator(
        task_id         = 'fail_deliberately',
        python_callable = fail_this_task
    )

    AlwaysUpstreamFailed = DummyOperator(
        task_id         = 'always_upstream_failed',
        trigger_rule    = TriggerRule.ALL_SUCCESS
    )

    VersionDependent = DummyOperator(
        task_id         = 'skipped_in_v1_but_upstream_failed_in_v2',
        trigger_rule    = TriggerRule.ONE_SUCCESS
    )

    FailHandler = DummyOperator(
        task_id         = 'this_handler_should_be_skipped',
        trigger_rule    = TriggerRule.NONE_SKIPPED
    )
    
    FailDeliberately >> AlwaysUpstreamFailed >> VersionDependent >> FailHandler

我的原始解决方案更复杂，处理程序有更多输入，即使来自始终处于upstream_failed 状态的任务，以及来自在两个 Airflow 版本中具有不同状态的任务之后的任务等 - 所以重构没那么容易。

【问题讨论】：

标签： python airflow airflow-2.x

【解决方案1】：

我认为您指的是PR，它改变了ONE_SUCCESS 的行为，当没有上游成功时，将任务标记为UPSTREAM_FAILED。

您描述的先前行为实际上是一个错误。当一项任务由于上游任务而失败时，这意味着某些事情是不正确的。另一方面，被跳过的任务被认为是一种预期的行为（分支运算符就是一个例子）。这也是 PR 面向 Airflow 2.1 的原因 - 这不是重大更改，这是一个错误修复。

【讨论】：