airflow - 如何增加每秒排队的任务？

Question

我正在尝试诊断性能不佳的气流管道，并想知道我应该从气流调度程序中获得什么样的性能，类似于“每秒调度的任务”。

我几乎没有排队的作业，而且我的许多任务在几秒钟内完成，所以我怀疑调度程序是限制组件，我有很多快速任务是我的错。不过，如果可以避免的话，我宁愿不重写我的 DAG。

我可以做些什么来提高调度程序排队任务的速率？

---

管道详细信息

这是我当前的airflow.cfg 的样子。

我只有两个 dags 在运行。一个每 5 分钟安排一次，另一个很少由第一个触发。我目前正试图以这种频率回填几年，但可能需要改变我的方法：

至于工作节点：我目前有 4 个相当强大的服务器运行，在磁盘、网络、cpu、RAM、交换方面的资源使用率不到 10%。关闭 3 个工作人员对我的任务吞吐量没有影响，并且服务器保持打开状态甚至几乎没有记录工作量的变化。

Answer 1

您的许多配置值airflow.cfg可能与此相关。

下[core]：

• 并行度：一次可以运行的任务实例总数。
• dag_concurrency：每次 DAG 运行可以运行的任务实例的限制，如果您有许多并行任务，可能需要增加。定义 DAG 时可以覆盖。
• non_pooled_task_slot_count：没有配置可以立即运行的池的任务限制。
• max_active_runs_per_dag：每个 DAG 运行的活动 DAG 的最大数量。如果您要手动触发运行，或者有 DAG 运行的备份计划以较短的时间间隔进行。定义 DAG 时可以覆盖。

下[scheduler]：

• schedule_heartbeat_sec：定义调度程序运行的频率，尝试使用较低的值。
• min_file_process_interval：每 N 秒最多处理一次每个文件。设置为 0 以从不限制处理文件的频率。

下[worker]：

• celeryd_concurrency： celery 将运行的工作人员数量，因此基本上一个工作人员可以一次执行的任务实例数。匹配 CPU 数量是一个流行的起点，但绝对可以更高。

最后一个是仅当您使用 时CeleryExecutor，如果您希望增加任务吞吐量，我绝对会推荐它。