使用复杂的触发规则构建 Airflow DAG

Question

这可能是一个逻辑问题，但它让我感到难过。我有以下数据：

我有两个主要的分支事件让我感到困惑：

在A之后，只有分支B或C应该运行。
在 B 之后，D 应该选择性地在 E.运行之前运行。

现在我已经实现了这个，所以 E 有触发规则“none_failed”，这是防止在跳过 D 时跳过 E 所必需的。此设置适用于所有情况，除非 C 需要运行。在这种情况下，C 和 E 运行同时（您可以通过方框看到颜色 B/D 被按预期跳过，但 E 是绿色）。阅读触发规则文档后，我明白了 为什么 会发生这种情况（两个 E 的父任务都被跳过，因此运行s）。但是我似乎无法弄清楚如何从中获得预期的行为（例如，当该分支为运行但 时保持 B/D/E 的当前行为 运行 E当C为运行).

关于其他上下文，这不是我的整个 DAG。任务 C 和 E 会聚到它们下游的另一个任务，触发规则为 ONE_FAILED，但为了简单起见，我在示例中省略了这一点。对如何获得预期行为有任何想法吗？

Answer 1

这可能不是最好的解决方案，但它似乎涵盖了您的所有场景。主要是我在E之前添加了一个虚拟任务以控制E的时间并将E的trigger_rule更改为“one_success”。

"one_success" 需要至少 1 个直接父级才能继承，因此对于 E，D 或 dummy 必须成功才能使 E 成为运行。

A = BranchPythonOperator(task_id='A', python_callable=_branch_A, dag=dag)

B = BranchPythonOperator(task_id='B', python_callable=_branch_B, dag=dag)
C = DummyOperator(task_id='C', dag=dag)

D = PythonOperator(task_id='D', python_callable=_test, dag=dag)
dummy = DummyOperator(task_id='dummy', dag=dag)
E = DummyOperator(task_id='E', trigger_rule='one_success', dag=dag)

A >> [B, C]
B >> [D, dummy] >> E

演示

使用复杂的触发规则构建 Airflow DAG

Structuring Airflow DAG with complex trigger rules

directed-acyclic-graphs

airflow