上游跳跃时气流“无故障”跳过

Question

我有一个工作流，其中我有两个并行进程（sentinel_run和sentinel_skip），它们应该根据条件运行或跳过，然后连接在一起（resolve）。我需要直接位于sentinel_任务下游的任务进行级联跳过，但是当它到达resolve任务时，resolve应该运行，除非上游的任一进程出现故障。在

根据documentation，“无故障”触发规则应该起作用：

none_failed: all parents have not failed (failed or upstream_failed) i.e. all parents have succeeded or been skipped

它也是一个related question的答案。在

然而，当我实现了一个小例子时，我看到的并不是这样：

from airflow.models import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import ShortCircuitOperator
from airflow.utils.dates import days_ago

dag = DAG(
    "testing",
    catchup=False,
    schedule_interval="30 12 * * *",
    default_args={
        "owner": "test@gmail.com",
        "start_date": days_ago(1),
        "catchup": False,
        "retries": 0
    }
)

start = DummyOperator(task_id="start", dag=dag)

sentinel_run = ShortCircuitOperator(task_id="sentinel_run", dag=dag, python_callable=lambda: True)
sentinel_skip = ShortCircuitOperator(task_id="sentinel_skip", dag=dag, python_callable=lambda: False)

a = DummyOperator(task_id="a", dag=dag)
b = DummyOperator(task_id="b", dag=dag)
c = DummyOperator(task_id="c", dag=dag)
d = DummyOperator(task_id="d", dag=dag)
e = DummyOperator(task_id="e", dag=dag)
f = DummyOperator(task_id="f", dag=dag)
g = DummyOperator(task_id="g", dag=dag)

resolve = DummyOperator(task_id="resolve", dag=dag, trigger_rule="none_failed")

start >> sentinel_run >> a >> b >> c >> resolve
start >> sentinel_skip >> d >> e >> f >> resolve

resolve >> g

此代码创建以下dag：

问题是resolved任务应该执行（因为上游没有upstream_failed或{}），但是它正在跳过。在

我已经反省了数据库，没有隐藏任何失败或上游失败的任务，我不明白为什么它不遵守“无失败”逻辑。在

我知道"ugly workaround"并在其他工作流中实现了它，但它增加了另一个要执行的任务，并增加了DAG的新用户必须摸索的复杂性（尤其是当您将其乘以多个任务时…）。这是我从气流1.8升级到气流1.10的主要原因，所以我希望我遗漏了一些明显的东西。。。在

Answer 1

这似乎是气流中的一个缺陷。如果要修复，请将您的声音添加到https://issues.apache.org/jira/browse/AIRFLOW-4453。在