Spark任务触发collect致结果超maxResultSize，求非调参解决方案

问题原因 #

从堆栈追踪可明确：spark.read.parquet()底层调用InMemoryFileIndex.bulkListLeafFiles方法时，会触发RDD.collect()将所有待读取文件的元数据（路径、大小、修改时间等）收集到Driver端。当传入的unifiedPathsFromAllStorages包含大量小文件或过多目录时，这些元数据的总大小会超出spark.driver.maxResultSize限制，从而引发报错。该collect操作是Spark构建文件索引的隐式逻辑，并非用户手动调用。

非增大maxResultSize的解决方案 #

1. 缩小扫描范围 #

• 清理unifiedPathsFromAllStorages中的冗余路径，仅保留业务需要的目录（如移除过期数据路径）。
• 利用数据分区特性（如日期、业务类型分区），直接指定分区路径或添加分区过滤条件，减少扫描文件数量：

  // 直接读取指定分区路径
  val telemetry = spark.read.parquet("path/to/data/date=2024-05-*")
  // 通过过滤条件下推实现分区裁剪
  val telemetry = spark.read.parquet("path/to/data").filter("date >= '2024-05-01'")
  

2. 合并小文件 #

针对Parquet存储中的大量小文件，提前合并为大文件，减少文件总数以降低元数据总量：

// 合并已有Parquet数据
spark.read.parquet("original/data/path")
  .repartition(8) // 根据集群资源合理设置分区数
  .write.mode("overwrite")
  .parquet("merged/data/path")

后续读取合并后的路径，可大幅降低元数据收集的压力。

3. 调整Spark文件读取配置 #

• 确认分区过滤下推开启：设置spark.sql.parquet.filterPushdown=true（Spark 2.4默认开启），确保分区过滤条件下推至文件扫描阶段，跳过无关目录。
• 利用Hive元数据（若数据已注册至Hive）：设置spark.sql.hive.convertMetastoreParquet=true，直接从Hive元数据获取文件列表，避免Spark全量扫描路径。
• 调整文件分区阈值：增大spark.sql.files.maxPartitionBytes（默认128MB），减少文件分区数，间接降低元数据总量。

4. 添加业务层过滤条件 #

如果业务无需全量数据，在读取时添加严格的谓词过滤，让Spark仅扫描符合条件的文件，无需收集所有文件元数据：

val telemetry = spark.read.parquet("path/to/data")
  .filter("machine_id = 'xxx' and timestamp >= '2024-05-01 00:00:00'")

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