BranchPythonOperator 后的 Airflow 任务不会失败并正确成功

Question

在我的 DAG 中，我有一些任务只能在运行星期六进行。因此，我使用 BranchPythonOperator 在星期六的任务和 DummyTask 之间进行分支。之后，我加入了两个分支并想运行其他任务。

工作流程如下所示：
在这里，我将 dummy3 的触发规则设置为 'one_success' 并且一切正常。

我遇到的问题是 BranchPythonOperator 上游的某些东西失败了：
BranchPythonOperator 和分支正确地具有状态 'upstream_failed'，但加入分支的任务变为 'skipped'，因此整个工作流显示 'success'.

我尝试使用 'all_success' 作为触发规则，然后如果出现问题，整个工作流程都会失败，但它会正常工作，但如果没有失败，则会跳过 dummy3。

我也试过'all_done'作为触发规则，然后如果没有失败它就可以正常工作，但是如果有什么失败dummy3仍然会被执行。

我的测试代码如下所示：

from datetime import datetime, date
from airflow import DAG
from airflow.operators.python_operator import BranchPythonOperator, PythonOperator
from airflow.operators.dummy_operator import DummyOperator

dag = DAG('test_branches',
          description='Test branches',
          catchup=False,
          schedule_interval='0 0 * * *',
          start_date=datetime(2018, 8, 1))


def python1():
    raise Exception('Test failure')
    # print 'Test success'


dummy1 = PythonOperator(
    task_id='python1',
    python_callable=python1,
    dag=dag
)


dummy2 = DummyOperator(
    task_id='dummy2',
    dag=dag
)


dummy3 = DummyOperator(
    task_id='dummy3',
    dag=dag,
    trigger_rule='one_success'
)


def is_saturday():
    if date.today().weekday() == 6:
        return 'dummy2'
    else:
        return 'today_is_not_saturday'


branch_on_saturday = BranchPythonOperator(
    task_id='branch_on_saturday',
    python_callable=is_saturday,
    dag=dag)


not_saturday = DummyOperator(
    task_id='today_is_not_saturday',
    dag=dag
)

dummy1 >> branch_on_saturday >> dummy2 >> dummy3
branch_on_saturday >> not_saturday >> dummy3

编辑

我刚刚想出了一个丑陋的解决方法：
dummy4 表示我实际需要执行的任务运行，dummy5 只是一个虚拟对象。
dummy3 仍然有触发规则 'one_success'.

现在 dummy3 和 dummy4 运行如果上游没有失败，dummy5 'runs' 如果那天不是星期六，如果那天是星期六则跳过，这意味着 DAG 被标记为成功在这两种情况下。
如果上游出现故障，则跳过 dummy3 和 dummy4，将 dummy5 标记为 'upstream_failed'，将 DAG 标记为失败。

此解决方法使我的 DAG 运行成为我想要的，但我仍然更喜欢没有一些 hacky 解决方法的解决方案。

Answer 1

您可以使用的一种解决方法是将 DAG 的第二部分放在 SubDAG 中，就像我在以下说明示例的代码中所做的那样：https://gist.github.com/cosenal/cbd38b13450b652291e655138baa1aba

它按预期工作，并且可以说它比您的解决方法更干净，因为您没有任何额外的辅助虚拟运算符。然而，您失去了平面结构，现在您必须放大 SubDag 才能看到内部的细节。

更一般的观察：在对您的 DAG 进行试验后，我得出结论，Airflow 需要类似 JoinOperator 的东西来替换您的 Dummy3 运算符。让我解释。您描述的行为来自这样一个事实，即 DAG 的成功仅基于最后一个运算符的成功（或跳过！）。

以下以“成功”状态结尾的 DAG 是支持上述声明的 MWE。

def python1():
    raise Exception('Test failure')

dummy1 = PythonOperator(
    task_id='python1',
    python_callable=python1,
    dag=dag
)

dummy2 = DummyOperator(
    task_id='dummy2',
    dag=dag,
    trigger_rule='one_success'
)

dummy1 >> dummy2

只有当直接父级之一成功并且所有其他父级都被跳过时才触发的 JoinOperator 会很酷，而不必使用 trigger_rule 争论。

或者，可以解决您遇到的问题的方法是触发规则 all (success | skipped)，您可以将其应用于 Dummy3。很遗憾，我认为您还不能在 Airflow 上创建自定义触发规则。

编辑：在这个答案的第一个版本中，我声称触发规则 one_success 和 all_success 根据 [=37] 的成功程度触发=]all DAG 中运算符的祖先，而不仅仅是直接父代。这与 documentation and in fact it is invalidated by the following experiment: https://gist.github.com/cosenal/b607825539aa0d308f10f3095e084fac

不匹配

Answer 2

将 dummy3 的触发规则设置为 'none_failed' 将使其在任何情况下都以预期状态结束。

参见 https://airflow.apache.org/concepts.html#trigger-rules

EDIT ：看起来这个 'none_failed' 触发规则在提出和回答这个问题时还不存在：它是在 2018 年 11 月添加的

见https://github.com/apache/airflow/pull/4182

BranchPythonOperator 后的 Airflow 任务不会失败并正确成功

Airflow task after BranchPythonOperator does not fail and succeed correctly

python

airflow

编辑