如何运行气流 dag 指定次数？
我尝试使用 TriggerDagRunOperator，该运算符对我有用。 在可调用函数中，我们可以检查状态并决定是否继续。
但是需要维护当前的计数和状态。
使用上述方法，我可以重复 DAG 'run'。
需要专家意见，有没有其他深刻的方法来运行 Airflow DAG X 次？ 谢谢。
【问题讨论】：
标签： airflow google-cloud-composer
恐怕 Airflow 完全是基于时间的调度。
您可以将计划设置为None，然后使用API to trigger runs，但您将在外部执行此操作，从而维护确定何时以及为何触发外部的计数和状态。
当您说您的 DAG 可能有 5 个任务要运行 10 次并且运行需要 2 个小时并且您无法根据时间安排它时，这会令人困惑。我们不知道 2 小时对您有什么意义，或者为什么它必须是 10 次运行，也不知道为什么您不能安排它每天运行一次这 5 个任务。使用简单的每日计划，它将每天在大约相同的时间运行一次，并且在任何一天花费的时间都不会超过 2 小时。对吧？
您可以将 start_date 设置为 11 天前（虽然是固定日期，但不要动态设置），将 end_date 设置为今天（也是固定的），然后添加每日 schedule_interval 和 @ 987654327@ of 1，您将获得 10 次运行，它会在相应地更改 execution_date 时将它们背靠背运行而不会重叠，然后停止。或者，您可以将 airflow backfill 与 None 计划的 DAG 和一系列执行日期时间一起使用。
您的意思是您希望它每 2 小时连续运行一次，但有时它会运行更长时间并且您不希望它重叠运行？好吧，您绝对可以安排它每 2 小时运行一次 (0 0/2 * * *) 并将 max_active_runs 设置为 1，这样如果前一次运行尚未完成，下一次运行将等待，然后在前一次运行完成时启动.请参阅https://airflow.apache.org/faq.html#why-isn-t-my-task-getting-scheduled 中的最后一个项目符号。
如果您希望您的 DAG 恰好每 2 小时运行一次 [给予或接受一些调度程序延迟，是的，这是一件事] 并让之前的运行继续进行，这主要是默认行为，但您可以添加 @987654333 @ 到一些本身不应该同时运行的重要任务（例如创建、插入或删除临时表），或使用具有单个插槽的池。
如果您的下一个计划已准备好开始，则没有任何功能可以终止先前的运行。如果之前的运行尚未完成，可能会跳过当前运行，但我忘记了具体是如何完成的。
这基本上是您的大部分选择。您也可以为计划外的 DAG 创建手动 dag_runs；一次创建 10 个（使用 UI 或 CLI 而不是 API，但 API 可能更容易）。
这些建议是否能解决您的顾虑？由于不清楚您为什么需要固定的运行次数、频率或使用什么计划和条件，因此很难提供具体的建议。
【讨论】：
BranchOperator 中结束您的 DAG，这将分支到 Dummy END 任务或 TriggerDagRunOperator 在其自己的 DAG id 和这会减少 Airflow 变量或其他一些外部数据源（DB、http get/put/post、S3/GCP 路径中的值等）以确定分支路径。
我们可以写一个自定义操作符/python 操作符
TaskInstance 表）运行的“n”个。 （请参阅task_command.py 寻求帮助）raise AirflowSkipException，reference）这篇优秀的文章可以启发灵感：Use apache airflow to run task exactly once
注意
这种方法的缺点是它假定任务的历史运行 (TaskInstances) 将永远保留（并且正确）
catchup 设置为 False）【讨论】：