python - 如何运行超过 10 万个任务的 Airflow dag？

Question

我有一个气流 DAG，它有超过 100,000 个任务。我最多只能运行 1000 个任务。除了调度程序挂起之外，网络服务器无法呈现任务并且在 UI 上非常慢。

我试过增加min_file_process_interval和processor_poll_interval配置参数。

我已设置num_duration为 3600，以便调度程序每小时重新启动。

我对网络服务器或调度程序有任何限制吗？一般来说，Airflow 中的大量任务如何处理？任何配置设置等都会非常有帮助。

另外，我是否应该以这种规模使用 SubDagOperator？请指教。

谢谢，

标签： pythonpython-3.xairflowairflow-scheduler

Answer 1

我能够运行超过 165,000 个气流任务！

但有一个问题。并非所有任务都在单个 Airflow Dag 中安排和呈现。

当我试图安排越来越多的任务时，我遇到的问题是调度程序和网络服务器的问题。随着越来越多的任务被调度，调度程序和网络服务器上的内存和 CPU 消耗急剧增加（这是显而易见的并且是有道理的）。它到了节点无法再处理它的地步（调度程序为 16,000 多个任务使用了超过 80GB 的内存）

我将单个 dag 分成 2 个 dag。一个是领导者/主人。第二个是工人dag。

我有一个气流变量，表示一次要处理多少个任务（例如，num_tasks=10,000）。由于我有超过 165,000 个任务，worker dag 将分 17 个批次一次处理 10k 个任务。

leader dag，它所做的只是用不同的 10k 任务集一遍又一遍地触发同一个 worker dag，并监控 worker dag 的运行状态。第一个触发器操作符为第一组 10k 个任务触发 worker dag，并一直等待直到 worker dag 完成。完成后，它会触发下一批 10k 任务的同一个 worker dag，依此类推。

这样，worker dag 不断被重用，并且永远不必调度超过 Xnum_tasks

最重要的是，找出max_number您的 Airflow 设置可以处理的任务。然后以领导者/工人的方式max_tasks一遍又一遍地启动 dag，直到完成所有任务。

希望这会有所帮助。