airflow - 如何运行 Airflow DAG 特定次数？

Question

如何运行指定次数的airflow dag？

我尝试使用 TriggerDagRunOperator，这个运算符适合我。在可调用函数中，我们可以检查状态并决定是否继续。

但是需要维护当前的计数和状态。

使用上述方法，我能够重复 DAG“运行”。

求高人指点，有没有其他高深的方法让Airflow DAG运行X次？谢谢。

Answer 1

恐怕 Airflow 完全是基于时间的调度。
您可以将计划设置为 None，然后使用 API to trigger runs ，但您会在外部执行此操作，从而维护确定何时以及为何在外部触发的计数和状态。

当您说您的 DAG 可能有 5 个任务，您想要运行 10 次并且一次运行需要 2 小时并且您无法根据时间安排它时，这很令人困惑。我们不知道 2 小时对您有什么意义，或者为什么必须运行 10 次，也不知道为什么您不能安排它每天运行一次这 5 个任务。如果使用简单的每日计划，它将每天大约在同一时间运行一次，并且在任何给定的一天花费多于 2 小时也没有关系。对吧？

您可以将 start_date 设置为 11 天前(虽然是固定日期，但不要动态设置)，将 end_date 设置为今天(也是固定的)和然后添加每日 schedule_interval 和 max_active_runs 为 1，您将得到恰好 10 次运行，并且它会在更改 execution_date 的同时连续运行它们而不会重叠 相应地，然后停止。或者您可以只使用 airflow backfill 和一个 None 计划的 DAG 和一系列执行日期时间。

你的意思是你想让它每2小时连续运行一次，但有时它会运行更长时间并且你不希望它重叠运行？好吧，您绝对可以安排它每 2 小时运行一次 (0 0/2 * * *) 并将 max_active_runs 设置为 1，这样如果之前的运行没有t finished 下一次运行将等待，然后在前一个运行完成时开始。请参阅 https://airflow.apache.org/faq.html#why-isn-t-my-task-getting-scheduled 中的最后一个项目符号.

如果您希望您的 DAG 恰好每 2 小时准时运行一次[给定或采取一些调度程序滞后，是的，这是一回事] 并且让之前的运行继续进行，这主要是默认行为，但您可以添加 depends_on_past 某些本身不应同时运行的重要任务(例如创建、插入或删除临时表)，或使用具有单个槽的池。

如果您的下一个计划已准备好开始，则没有任何功能可以终止之前的运行。如果之前的运行尚未完成，则可能会跳过当前运行，但我忘记了具体是如何完成的。

这基本上是您的大部分选择。您还可以为未安排的 DAG 创建手动 dag_run；一次创建 10 个(使用 UI 或 CLI 而不是 API，但 API 可能更容易)。

这些建议是否解决了您的顾虑？由于不清楚为什么要运行固定次数、运行频率或时间表和条件，因此很难提供具体建议。