Dask调度器读取大量Avro文件时出现OOM问题如何解决

问题结论 #

你当前的超大量文件读取方式属于典型的参数配置错误导致的反模式，不需要手动拆分Avro文件，调整读取参数即可解决问题。

根因分析 #

你设置的blocksize=5000000仅为5MB，按你总数据规模计算：
总数据量 = 5300个 × 300MB/个 ≈ 1.59TB
按5MB拆分后将生成超过30万个分区任务，Dask调度器需要为每个任务维护元数据、依赖关系、执行状态，所有任务生成完成前不会下发给工作节点，这就是调度器CPU占满、内存溢出，而worker全程空闲的核心原因。

内置优化方案 #

直接调整读取参数即可解决，无需修改源文件：

• 调大blocksize到合理区间：Dask处理云存储批量文件的最优blocksize范围为64MB~256MB，你可以直接设置为128MB，调整后总分区数会降到约12400个，64GB内存的调度节点完全可以支撑。
  优化后的读取代码：

  bag = db.read_avro(
      'gcs://mybucket/myfiles-*.avro',
      blocksize=128 * 1024 * 1024
  )
  

• 开启任务融合优化：如果调整blocksize后仍有调度压力，可以在读取时添加fused=True参数（Dask 2021.06及以上版本的read_avro原生支持），将多个相邻的小任务合并为单个大任务，进一步降低调度器开销。

额外优化建议 #

• 转存Parquet前可以根据你的集群worker总核数做一次重分区，分区数设置为总核数的2~3倍即可，可进一步提升执行效率，减少写入Parquet时的小文件问题。
• Avro本身是可拆分格式，Dask原生支持单Avro文件切分、多小Avro文件合并到同一分区，完全不需要手动拆分源文件。

内容的提问来源于stack exchange，提问作者sobek