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

Question

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

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

如何提高调度程序对任务进行排队的速率？

---

管道详情

Here is what my current airflow.cfg looks like.

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

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

Answer 1

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

[core]下：

• parallelism：一次可以运行的任务实例总数。
• 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 时，如果您希望提高任务吞吐量，我绝对会推荐它。