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 会“运行”；如果当天是星期六，则会跳过，这意味着 DAG 在这两种情况下都被标记为成功。
如果上游出现故障，dummy3 和 dummy4 会被跳过，dummy5 会被标记为，'upstream_failed'并且 DAG 会被标记为失败。

此解决方法使我的 DAG 按我希望的方式运行，但我仍然更喜欢没有一些 hacky 解决方法的解决方案。

Answer 1

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

\n\n

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

\n\n\n\n

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

\n\n

以下 DAG 以 \xc2\xabSuccess\xc2\xbb 状态结尾，是支持上述声明的 MWE。

\n\n

def python1():\n    raise Exception(\'Test failure\')\n\ndummy1 = PythonOperator(\n    task_id=\'python1\',\n    python_callable=python1,\n    dag=dag\n)\n\ndummy2 = DummyOperator(\n    task_id=\'dummy2\',\n    dag=dag,\n    trigger_rule=\'one_success\'\n)\n\ndummy1 >> dummy2\n

\n\n

\n\n

如果有一个 JoinOperator 仅在其中一个直接父级成功并且所有其他父级都被跳过而无需使用trigger_rule参数时才触发，那将是很酷的。

\n\n

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

\n\n

编辑：在这个答案的第一个版本中，我声称触发规则one_success和all_success触发取决于DAG 中操作员的所有祖先的成功程度，而不仅仅是直接父母。这与文档不符，事实上，以下实验使其无效：https ://gist.github.com/cosenal/b607825539aa0d308f10f3095e084fac

\n\n

\n