Spark DataFrame大表全外连接OOM问题及替代方案求助

碰到这种大规模全外连接OOM的情况太常见了，尤其是两个超大型DataFrame的时候。结合你的集群配置（50节点、8GB executor内存），我给你几个经过生产环境验证的可行方案，按优先级排序：

1. 先排查数据倾斜，优化shuffle核心参数 #

全外连接OOM的头号元凶通常是数据倾斜——如果某些A+B组合的行数特别多，会导致单个executor要处理远超内存承载的数据集。另外默认的shuffle配置可能不匹配你的数据规模：

• 先统计连接键的分布，确认是否有倾斜：

  // 统计DF1中每个A+B组合的行数
  df1.groupBy("A", "B").count().orderBy(desc("count")).show(20)
  // DF2同理
  df2.groupBy("A", "B").count().orderBy(desc("count")).show(20)
  

  如果发现有Top N的组合行数远超其他，先对这些倾斜键做加盐处理：给倾斜的A+B加上随机后缀（比如0-9的数字），分成小批次join后再合并，避免单个分区过载。
• 调整shuffle分区数：默认的spark.sql.shuffle.partitions是200，对5.39亿行的数据来说太少了。建议设置为集群总core数的2-3倍（比如50节点，每个节点4core的话，设置为50*4*2=400，或者更大的1000-2000），让数据更均匀地分布到各个executor：

  spark.conf.set("spark.sql.shuffle.partitions", "2000")
  

• 增大executor内存开销：shuffle过程会用到堆外内存，默认的spark.executor.memoryOverhead是executor内存的10%（8GB的话只有800M），很容易不够。建议设置为2GB：

  spark.conf.set("spark.executor.memoryOverhead", "2g")
  

2. 按连接键预分区并持久化到磁盘 #

先对两个DF按A+B进行分区，然后持久化到磁盘（避免内存不足），这样后续join时不需要再全局shuffle，只需要在每个节点内处理对应分区的数据：

import org.apache.spark.storage.StorageLevel

// 按连接键重分区，确保相同A+B的在同一个分区
val df1Partitioned = df1.repartition(col("A"), col("B")).persist(StorageLevel.DISK_ONLY)
val df2Partitioned = df2.repartition(col("A"), col("B")).persist(StorageLevel.DISK_ONLY)

// 触发持久化（可选，提前计算分区数据）
df1Partitioned.count()
df2Partitioned.count()

// 执行全外连接
val joinedDf = df1Partitioned.join(df2Partitioned, Seq("A", "B"), "fullouter")

这种方式能大幅减少shuffle的数据量和内存占用，因为数据已经按连接键分布到对应节点了。

3. 改用RDD API做细粒度控制 #

如果DataFrame的自动优化还是搞不定，可以转成RDD，更灵活地控制分区、存储和join逻辑：

import org.apache.spark.sql.types._
import org.apache.spark.storage.StorageLevel

// 转成RDD，以(A,B)为key
val rdd1 = df1.rdd.map(row => 
  ((row.getAs[String]("A"), row.getAs[String]("B")), 
   (row.getAs[Double]("C"), row.getAs[Double]("D")))
)
val rdd2 = df2.rdd.map(row => 
  ((row.getAs[String]("A"), row.getAs[String]("B")), 
   (row.getAs[Double]("E"), row.getAs[Double]("F")))
)

// 设置合适的分区数，执行全外连接并持久化到磁盘
val joinedRdd = rdd1.fullOuterJoin(rdd2, numPartitions = 2000).persist(StorageLevel.DISK_ONLY)

// 转回DataFrame
val joinedDf = spark.createDataFrame(joinedRdd.map { 
  case ((a, b), (cdOpt, efOpt)) => 
    (a, b, 
     cdOpt.map(_._1).getOrElse(null), 
     cdOpt.map(_._2).getOrElse(null), 
     efOpt.map(_._1).getOrElse(null), 
     efOpt.map(_._2).getOrElse(null))
}, StructType(Seq(
  StructField("A", StringType),
  StructField("B", StringType),
  StructField("C", DoubleType),
  StructField("D", DoubleType),
  StructField("E", DoubleType),
  StructField("F", DoubleType)
)))

RDD的join能让你更精准地控制每个步骤的内存使用，避免DataFrame API隐藏的自动操作带来的意外。

4. 分批次处理（业务允许的情况下） #

如果不需要一次性生成完整结果，可以把数据按A列拆分多个批次，逐个处理后合并结果：

// 获取所有唯一的A值，分成小批次
val uniqueAValues = df1.select("A").distinct().collect().map(_.getAs[String]("A"))
val batches = uniqueAValues.grouped(100000).toList // 每个批次10万个A值

// 逐个批次执行join，收集结果
val resultDfs = batches.map { batch =>
  val df1Batch = df1.filter(col("A").isin(batch:_*))
  val df2Batch = df2.filter(col("A").isin(batch:_*))
  df1Batch.join(df2Batch, Seq("A", "B"), "fullouter")
}

// 合并所有批次结果
val finalDf = resultDfs.reduce(_ union _)

这种方法能把单批次的数据量降到原来的1/N，大幅降低内存压力，但要注意如果某些A对应的行数特别多，还是要结合数据倾斜处理。

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