Spark使用wholeTextFile处理小文件内存过高的优化问询

替换wholeTextFile为流式处理 #

wholeTextFile会将整个压缩文件解压后加载为单个字符串（单文件解压后100MB），直接导致单分区内存占用过高；且GZ是不可分割压缩格式，设置minPartitions=400不会生效（每个GZ文件只能对应1个分区）。改用textFile逐行读取，结合mapPartitions实现流式分割：

val rdd = sc.textFile("path/to/your/*.gz")
val splitRdd = rdd.mapPartitions(iter => {
  val contentBuffer = new StringBuilder()
  // 逐行处理，累积内容直到遇到分隔符
  val intermediate = iter.flatMap(line => {
    if (line.startsWith("-+")) { // 匹配分隔符行（根据实际分隔符调整判断逻辑）
      val block = contentBuffer.toString().trim
      contentBuffer.clear()
      if (block.nonEmpty) Some(block) else None
    } else {
      contentBuffer.append(line).append("\n")
      None
    }
  })
  // 处理最后一个未被分隔符截断的内容块
  val finalBlock = contentBuffer.toString().trim
  if (finalBlock.nonEmpty) intermediate ++ Iterator(finalBlock) else intermediate
})
// 转换为DataFrame并写入Parquet
splitRdd.toDF("content").write.mode("overwrite").parquet("path/to/output")

此方式避免一次性加载整个文件到内存，单个分区内存占用仅为当前累积的内容块大小。

避免不必要的内存持有 #

• 不要将中间RDD赋值给全局变量，处理完成后让其失去引用，便于GC回收。
• 转换为DataFrame时，仅保留需要的字段，避免冗余数据。

本地模式下Driver与Executor共享同一JVM进程，需重点优化Driver内存及GC配置：

val conf = new SparkConf()
  .setMaster("local[16]") // 匹配机器核心数，最大化并行度
  .setAppName("GzToParquet")
  .set("spark.driver.memory", "40g") // 分配40G内存（预留20+G给系统及其他进程）
  .set("spark.memory.fraction", "0.8") // 80%内存用于计算与存储
  .set("spark.memory.storageFraction", "0.2") // 仅20%内存用于缓存，更多留给计算
  .set("spark.driver.maxResultSize", "0") // 取消结果大小限制（写入Parquet无需返回结果到Driver）
  .set("spark.default.parallelism", "32") // 设置默认并行度为核心数的2倍
  .set("spark.sql.shuffle.partitions", "32") // 减少Shuffle分区数，降低内存开销
  .set("spark.driver.extraJavaOptions", "-XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:+HeapDumpOnOutOfMemoryError")
val sc = new SparkContext(conf)

• G1GC垃圾回收器：更适合大内存场景，能有效控制GC停顿时间，减少内存碎片。
• 内存比例调整：优先保障计算内存，因为任务以数据处理为主，无需大量缓存。

• 任务完成后显式调用sc.stop()关闭Spark上下文，释放所有关联资源。
• 避免在Driver端执行collect()、count()等会将大量数据拉取到本地的操作（若需统计，改用rdd.count()而非拉取全量数据）。
• 若仍存在内存未释放，可在任务结束前添加System.gc()（仅作为最后手段，优先依赖代码与配置优化）。

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