在我的DAG中,我有一些任务应该只在星期六运行。因此,我使用branchpython运算符在周六的任务和DummyTask之间进行分支。之后,我加入两个分支,并希望运行其他任务。

工作流如下所示:
在这里,我将dummy3的触发规则设置为 'one_success' 一切都很好。

我遇到的问题是当branchpythonooperator的上游发生故障时:
分支运算符和分支的状态正确 '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仍然有触发规则 “一个成功” .

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

这个解决方案可以让我的DAG按我所希望的方式运行,但我仍然希望没有一些复杂的解决方案。