气流:如何重新运行相关DAG

Question

假设我有以下DAG依赖项：

/> DAG B -->....
/
/
DAG A ---> DAG C -->...


> DAG D -->...

这些DAG每天运行，DAG A的输出分别由DAG B、C、D消耗。

如果DAG A失败(例如，Spark作业有一个错误，我们想在修复后重新执行它(，如何确保我们也重新运行DAG B、C、D等，以便现在它们以固定的DAG A输出运行？它能自动完成吗？

Answer 1

您可以有一个单独的Main DAG(controller_DAG(，并使用TriggerDagRunOperator来触发不同的DAG。

使用TriggerDagRunOperator的任务之间的依赖关系可以通过以下方式设置：

dag_a >> [dag_b, dag_c, dag_d] >> ...

Answer 2

• 您可以为此使用TriggerDagRunOperatorhttps://www.astronomer.io/guides/trigger-dag-operator/

• 您也可以考虑使用SubDAGOperator将这些DAG合并为一个DAG。在这种情况下，子DAG b、c、d可以依赖于子DAG a。