读取大量Avro文件时Spark Driver内存溢出问题排查求助

核心原因排查 #

1. 文件元数据过载：即使没有执行collect，Spark读取大量文件时，Driver需要加载所有文件的路径、大小、分区信息等元数据。如果S3子目录下存在数万甚至数十万文件，元数据累积量会直接占满Driver内存。
2. Avro复杂Schema的隐性开销：即使预先指定了Schema，后续操作中涉及嵌套结构展开、类型转换等逻辑时，Driver需要维护额外的执行计划元数据，部分优化阶段（如谓词下推、分区裁剪）也会消耗大量内存。
3. S3对象存储的API特性：Spark列举S3文件时需逐个调用API，大量文件会导致Driver处理海量API返回结果，持续占用内存。

具体解决方案 #

1. 减少Driver需处理的元数据量 #

• 合并小文件：如果S3上存在大量小Avro文件，先通过Databricks将数据repartition后重新写入S3，或者用S3批量工具合并文件。建议每个文件大小设置为spark.sql.files.maxPartitionBytes的1-2倍（默认128MB），既减少文件数量，又保证分区合理性。
• 精准过滤读取路径：如果数据按目录分区（如year=2024/month=05），直接指定分区路径读取，避免扫描所有子目录：

  spark.read.format("avro").load("s3://bucket/path/year=2024/month=05")
  

  也可通过pathGlobFilter过滤文件类型，缩小扫描范围：

  spark.read.format("avro").option("pathGlobFilter", "*.avro").load("s3://bucket/path")
  

2. 调整Spark参数优化Driver内存 #

• 限制S3文件列举的单次返回量：设置spark.hadoop.fs.s3a.list.max-items，控制每次S3 API返回的文件数，避免一次性加载过多元数据：

  spark.conf.set("spark.hadoop.fs.s3a.list.max-items", "5000")
  

  注意该值不宜过大，否则单次请求返回的数据仍会压垮Driver。
• 关闭分区列类型自动推断：禁用spark.sql.sources.partitionColumnTypeInference.enabled，避免推断分区列类型时消耗额外内存：

  spark.conf.set("spark.sql.sources.partitionColumnTypeInference.enabled", "false")
  

• 配置Driver内存预留：除了spark.driver.memory，调整spark.driver.memoryOverhead为Driver预留额外内存，用于处理元数据和JVM开销：

  spark.conf.set("spark.driver.memoryOverhead", "4g")
  

3. 优化读取与后续执行逻辑 #

• 分批次读取处理：如果无法合并文件，分批次读取不同子目录，处理完一批再加载下一批，避免一次性加载全量元数据：

  val subDirs = List("s3://bucket/path/dir1", "s3://bucket/path/dir2")
  subDirs.foreach(dir => {
    val df = spark.read.format("avro").schema(preDefinedSchema).load(dir)
    // 执行后续处理逻辑（如写入目标表）
    df.write.mode("append").saveAsTable("target_table")
  })
  

• 规避不必要的宽依赖操作：后续操作中若有join、groupBy等宽依赖逻辑，先过滤冗余数据减少shuffle量，同时调整spark.sql.shuffle.partitions为集群核心数的2-3倍（默认200），避免生成过多中间文件增加Driver元数据维护压力。
• 转换为列式存储格式：将读取的Avro数据转换为Delta Lake格式存储，Delta会自动优化文件大小与分区，大幅降低后续操作的元数据压力。

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