python - 在任务成功/失败时重新安排 DAG

Question

考虑一个非常简单的 Apache Airflow DAG：

FileSensor -> PythonOperator

在哪里FileSensor等待一些文件出现（相对较短poke_interval）并PythonOperator处理这些文件。此 DAG 计划@once无限期运行 - 我如何将其设置为重新安排以PythonOperator在它成功（或失败）之后再次运行？

标签： pythonairflow-schedulerairflow

Answer 1

总的来说，我认为 Elad 的建议可能会奏效，但我认为这是一种不好的做法。DAG 在设计（和名称）上是非循环的，因此在其中创建任何类型的循环都可能导致其行为异常。

同样基于 Airflow 文档，如果您打算使用外部 dag 触发器，则应将 dag 计划设置为 None。就我个人而言，我不确定它是否一定会破坏某些东西，但它肯定会给你带来你意想不到的输出。如果出现问题，您稍后可能会花费更长的时间来调试它。

恕我直言，更好的方法是让您尝试并重新考虑您的设计。如果您需要在失败时重新安排 dag，您可以利用传感器的重新安排模式https://www.astronomer.io/guides/what-is-a-sensor。不知道为什么要在成功时重新运行它，如果源中有多个文件的情况，我会说宁愿在你的 dag 脚本中创建具有可变参数和 for 循环的多个传感器。