时间戳

首页 > 解决方案 > 大量任务时，气流调度程序不调度（或缓慢）

airflow-scheduler - 大量任务时，气流调度程序不调度（或缓慢）

问题描述

我正在使用 Google Cloud Composer（版本：composer-1.10.2-airflow-1.10.6）上的气流。

我意识到当有很多任务要处理时，调度程序不会调度任务（参见下面的甘特图）

甘特图 （不要注意颜色，红色任务是“createTable Operators”，如果表已经存在则失败，因此它们必须在 DAG 的下一部分（重要的部分）运行之前失败 5 次）

任务之间有几个小时的差距！（例如上午 10 点到下午 15 点之间的 5 小时，但什么也没发生）

通常它适用于大约 40 个 DAG，每个 DAG 大约有 100-200 个任务（有时更多）。但最近我添加了 2 个有很多任务的 DAG（每个约 5000 个），调度程序非常慢或不调度任务。在屏幕截图中，我在下午 15 点暂停了 2 个有很多任务的 DAG，调度程序又回来了，工作正常。

你有什么解决办法吗？

Airflow 旨在成为处理“无限”数量任务的工具。

以下是有关我的环境的一些信息：

以下是有关气流配置的一些信息：

╔════════════════════════════════╦═══════╗
║ Airflow parameter              ║ value ║
╠════════════════════════════════╬═══════╣
║ -(celery)-                     ║       ║
║ worker_concurrency             ║ 32    ║
║ -(webserver)-                  ║       ║
║ default_dag_run_display_number ║ 2     ║
║ workers                        ║ 2     ║
║ worker_refresh_interval        ║ 60    ║
║ -(core)-                       ║       ║
║ max_active_runs_per_dag        ║ 1     ║
║ dagbag_import_timeout          ║ 600   ║
║ parallelism                    ║ 200   ║
║ min_file_process_interval      ║ 60    ║
║ -(scheduler)-                  ║       ║
║ processor_poll_interval        ║ 5     ║
║ max_threads                    ║ 2     ║
╚════════════════════════════════╩═══════╝

谢谢您的帮助

编辑：

我的 26 个 DAG 是由单个 .py 文件通过解析一个巨大的 JSON 变量来创建的，以创建所有 DAG 和任务。

可能问题出在这里，因为今天 Airflow 正在调度来自其他 DAG 的任务，而不是我描述的 26 个（尤其是 2 个大 DAG）。更准确地说，Airflow 有时会调度我的 26 个 DAG 的任务，但它更容易和更频繁地调度其他 DAG 的任务。

标签： airflow-schedulergoogle-cloud-composerairflow

解决方案

高任务间延迟通常表明存在与调度程序相关的瓶颈（而不是与工作人员相关的瓶颈）。即使一遍又一遍地运行相同的 DAG，Composer 环境仍然可能会遇到这样的性能瓶颈，因为每次工作的分布可能不同，或者后台运行的进程可能不同。

首先，我建议增加调度程序可用的线程数 ( scheduler.max_threads)，然后确保您的调度程序不会消耗它所在节点的所有 CPU。您可以通过确定调度程序所在节点的位置，然后在 Cloud Console 中检查来检查调度程序所在节点的 CPU 指标。要查找节点名称：

# Obtain the Composer namespace name
kubectl get namespaces | grep composer

# Check for the scheduler
kubectl get pods -n $NAMESPACE -o wide | grep scheduler

如果上述方法没有帮助，那么调度程序也有可能故意阻塞某个条件。要检查调度程序检查要运行的任务时评估的所有条件，请设置core.logging_level=DEBUG. 在调度程序日志（您可以在 Cloud Logging 中对其进行过滤）中，您可以检查所有通过或失败的条件，以便任务运行或保持排队。

推荐阅读