Spark中对比两个10PB级大DataFrame内存溢出问题求优化方案

当前方法的核心问题 #

你用的行哈希生成+集合差+join的方式，在10PB量级下必然触发OOM：集合差（except）操作会触发全量shuffle，把两个DF的所有row_hash都拉到一起做全局对比，单节点内存根本扛不住这么大的数据量。

可行优化方案 #

1. 分桶局部对比 #

提前对两个DF按业务主键或高频区分列做分桶存储，分桶数量匹配集群executor数（比如1000个executor就设1000个分桶），让相同分桶的数据落在同一节点，把全局对比拆成每个分桶内的局部对比：

// 先将两个DF分桶存储（推荐Parquet/Orc格式）
df.write.bucketBy(1000, "business_key").saveAsTable("bucketed_df")
df1.write.bucketBy(1000, "business_key").saveAsTable("bucketed_df1")

// 分桶内执行对比，避免全量shuffle
val dfBucketed = spark.table("bucketed_df")
val df1Bucketed = spark.table("bucketed_df1")

// left_anti直接筛选出df有但df1没有的记录
val diffResult = dfBucketed.join(df1Bucketed, Seq("business_key", "row_hash"), "left_anti")

分桶后Spark会自动对齐同分区数据，shuffle量骤降，从根源避免内存溢出。

2. 分区增量处理 #

如果DF是按时间、地域等维度分区的（比如dt分区），直接按分区逐个对比，处理完一个分区就写入结果、释放内存：

// 获取所有分区列表
val partitionList = df.select("dt").distinct().collect().map(_.getString(0))

partitionList.foreach { partition =>
  // 加载单个分区数据
  val dfPart = df.filter(s"dt = '$partition'")
  val df1Part = df1.filter(s"dt = '$partition'")
  
  // 单分区内执行对比
  val diffPart = dfPart.select("row_hash").except(df1Part.select("row_hash"))
  val resultPart = diffPart.join(dfPart, Seq("row_hash"), "inner")
  
  // 结果直接写入外部存储，不累积在内存
  resultPart.write.mode("append").parquet("/path/to/diff_results")
}

注意：绝对不要把所有分区结果collect到driver，全程用分布式存储承接结果。

3. 多级哈希粗筛+细查 #

不要生成单一全局row_hash，改成按列分组生成多级哈希，先通过主键哈希粗筛掉匹配数据，再用明细哈希做精准对比：

import org.apache.spark.sql.functions._

// 生成两级哈希：主键哈希做粗筛，明细哈希做精准对比
val dfWithHashes = df.withColumn("primary_hash", hash("id", "order_id"))
                     .withColumn("detail_hash", hash("col_a", "col_b", "col_c"))
val df1WithHashes = df1.withColumn("primary_hash", hash("id", "order_id"))
                       .withColumn("detail_hash", hash("col_a", "col_b", "col_c"))

// 先按primary_hash过滤，再对比detail_hash
val diffResult = dfWithHashes.join(df1WithHashes, Seq("primary_hash"), "left_anti")
                           .filter(dfWithHashes("detail_hash") =!= df1WithHashes("detail_hash"))

这种方式能大幅减少join时的数据量，只处理真正的差异候选。

4. 核心参数调优 #

除了内存，重点优化shuffle和序列化：

• spark.sql.shuffle.partitions：设为executor数量的2-3倍，避免单个shuffle分区数据过大（比如100个executor就设300）。
• 开启Kyro序列化：spark.serializer org.apache.spark.serializer.KryoSerializer，比Java序列化内存占用少、速度快。
• 调整内存预留：spark.executor.memoryOverhead设为executor内存的20%-30%，给shuffle和系统进程留足内存。

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