如何避免Spark中昂贵操作重复执行及优化分区策略

• 应用流程：

  1. 按20个分区读取数据库，数据量有数百万行，存储格式为Blob，大小不确定
  2. 在mapPartition中遍历记录，将Blob转换为JSON表格格式（该操作开销极高）
  3. 在foreachPartition中使用累加器，通过JSON字符串长度计算每个分区的大小（注：步骤2中使用累加器始终返回0，因此移至后续foreachPartition执行）
  4. 若分区数据大小超过1GB，则重新分区
  5. 将JSON表格数据保存为磁盘CSV文件

• 当前核心问题：
  步骤2的Blob转JSON操作被执行了两次——一次是foreachPartition计算分区大小时，另一次是保存CSV时。需要优化流程避免重复执行；同时希望了解是否可以不进行重新分区，直接在现有20个分区的Executor核心中按大小拆分数据集并保存CSV文件。

一、避免重复执行Blob转JSON操作 #

Spark中转换算子（如mapPartition）是懒加载的，每次触发行动算子（如foreachPartition、save）都会重新执行整个依赖链，这就是重复执行昂贵操作的根源。解决办法很直接——持久化转换后的数据集：

1. 在步骤2完成mapPartition转换后，对生成的RDD/DataFrame调用persist()或cache()方法，将结果持久化到内存+磁盘（推荐使用StorageLevel.MEMORY_AND_DISK_SER，序列化存储可节省空间）。
   • Scala示例代码：

     val jsonDF = rawBlobDF.mapPartition(iter => {
       // 开销极高的Blob转JSON逻辑
       iter.map(blob => convertBlobToJson(blob))
     }).persist(StorageLevel.MEMORY_AND_DISK_SER)
     

2. 后续执行foreachPartition计算分区大小时，会直接读取持久化后的数据集，不会重新执行Blob转换；保存CSV时同样复用持久化数据，彻底避免重复计算。
3. 注意：完成所有操作后，记得调用unpersist()释放存储资源，避免占用过多集群空间。

二、避免重新分区，直接在现有分区内拆分保存 #

完全可以实现，核心思路是在单个分区内部按大小拆分数据，生成多个输出文件，无需重新分区。具体两种实现方式：

方案1：针对RDD自定义输出逻辑 #

在foreachPartition中遍历分区内的JSON数据时，实时累计数据大小，每当达到1GB阈值就写入新的CSV文件：

jsonRDD.foreachPartition(iter => {
  var currentSize = 0L
  var fileIndex = 0
  val maxSize = 1024 * 1024 * 1024 // 1GB
  var writer: CSVWriter = null

  try {
    iter.foreach(jsonRecord => {
      val recordStr = jsonRecord.toString()
      val recordSize = recordStr.getBytes(StandardCharsets.UTF_8).length

      if (writer == null || currentSize + recordSize > maxSize) {
        // 关闭当前writer（若存在）
        if (writer != null) {
          writer.close()
        }
        // 生成包含分区ID和文件索引的文件名
        val partitionId = TaskContext.getPartitionId()
        val fileName = s"output/part-${partitionId}-${fileIndex}.csv"
        writer = new CSVWriter(new FileWriter(fileName))
        fileIndex += 1
        currentSize = 0
      }

      writer.writeNext(convertJsonToCsvFields(recordStr))
      currentSize += recordSize
    })
  } finally {
    // 最后确保关闭writer
    if (writer != null) {
      writer.close()
    }
  }
})

这种方式完全复用原有20个分区的Executor核心，每个分区内部自行拆分文件，无需触发重新分区的 shuffle 操作。

方案2：针对DataFrame的优化 #

如果使用DataFrame API，最简单的方式是先将DataFrame转换为RDD，再套用上述自定义输出逻辑。另外Spark提供的maxRecordsPerFile参数可按记录数拆分文件，若能估算单条记录的平均大小，也可间接实现近似的大小拆分，但精度不如直接计算实际数据大小。

• 用持久化（persist）解决重复执行昂贵转换的问题，是最直接有效的方案；
• 可以避免重新分区，通过在现有分区内部按大小拆分数据并写入多个文件，充分利用原有Executor核心资源。

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