gpt4 book ai didi

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

转载 作者:太空狗 更新时间:2023-10-30 02:24:47 32 4
gpt4 key购买 nike

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

工作流程如下所示: enter image description here
在这里,我将 dummy3 的触发规则设置为 'one_success' 并且一切正常。

我遇到的问题是当 BranchPythonOperator 上游的某些东西失败时: enter image description here
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

编辑

我刚刚想出了一个丑陋的解决方法: enter image description here
dummy4 代表我实际需要运行的任务,dummy5 只是一个虚拟对象。
dummy3 仍然有触发规则 'one_success'

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

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

最佳答案

您可以使用的一种解决方法是将 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_successall_success 根据成功的程度触发 所有 DAG 中运算符的祖先,而不仅仅是直接父代。这与 documentation 不匹配事实上,它在以下实验中无效:https://gist.github.com/cosenal/b607825539aa0d308f10f3095e084fac

关于python - BranchPythonOperator 后的 Airflow 任务不会失败并正确成功,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/51664755/

32 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com