Airflow DAG成功回调

Question

有没有一种优雅的方式来定义DAG成功事件的回调函数？ 我真的不想设置一个任务，它将是所有其他任务的上游，带有on_success_callback。 谢谢！

- Aviv Oron

---

为什么不使用on_success_callback设置任务呢？我认为这在某种程度上很“优雅”。 :) - MoreFreeze

因为我必须将此任务配置为流程中所有“最终”任务的上游，以确保DAG中的所有任务都成功结束。我真的不想这样做，这会导致错误出现（比如忘记将某些“最终”任务作为该任务的父任务）。 - Aviv Oron

只有在1.9.0版本中，Operator才具有“on_success_callback”，因此任何解决方案都必须与Operator相关。您的问题可能是如何轻松创建这样的DAG。您可以编写一个装饰器或其他东西来实现此目的。 - MoreFreeze

Answer 1

如果我理解正确，您的DAG的最后一步是在成功时回调到其他系统。因此，我鼓励您精确地建模您的DAG。为什么要试图从DAG逻辑中隐藏该部分呢？这正是上游/下游建模的用途。出于美学考虑隐藏DAG逻辑的一部分听起来对我来说不是一个好的权衡。

我建议您将回调任务设为DAG中最终任务的下游而不是所有任务的下游。

- benjamin

Answer 2

不知道现在对你有多大帮助，但你要找的功能在Airflow 1.10中存在...一个DAG具有on_success_callback

https://airflow.apache.org/code.html#airflow.models.DAG

- user3027497

Answer 3

创建任务并以编程方式设置上游，这样你就不会忘记添加一个了，只要这是在文件的最后完成。

for task in dag.tasks:
    final_task.set_upstream(task)
# Only run if all the other tasks succeeded. EDIT: This is the default
final_task.trigger_rule = 'all_success'
dag.add_task(final_task)

- Ash Berlin-Taylor

---

我试图避免这样做，因为这样我的图表会非常丑 :( - Aviv Oron

你可以尝试避免它，但我不知道其他的方法 - DAGs没有任何钩子 - 钩子存在于任务中。 - Ash Berlin-Taylor

这样做不起作用。根据文档：https://airflow.incubator.apache.org/concepts.html#trigger-rules，“all_success”是默认值，意思是“如果所有父任务都成功，则触发”。 - benjamin

OP要求“DAG成功事件的回调”。听起来就是这样。 - Ash Berlin-Taylor