Spark DataFrame Join堆内存不足与分区过多问题求助

嘿，我完全懂你这种崩溃的心情——明明本地测试（22列）跑得飞快，到了集群上（190列），明明每个DF过滤后只有4行，Join完分区数直接飙到4万，任务要么慢得离谱要么直接崩。核心问题其实是过滤后的数据量极小，但你的代码继承了原始大DF的大量分区，再加上190列的宽表，每个小分区的调度、序列化开销被无限放大，最终压垮了集群资源。

给你几个针对性的解决方向，亲测有效：

1. 过滤后立刻合并分区，斩断原始分区的“尾巴” #

既然每个DF过滤后只剩4行，完全没必要保留原始的几千上万个分区。在过滤后直接用coalesce（无shuffle，更高效）把分区合并成1个：

val df1 = mainDf1.filter("metricname = 'NPV'")
                 .withColumnRenamed("my_attr","df1attr")
                 .withColumnRenamed("value","df1val")
                 .coalesce(1) // 合并为单个分区
val df2 = mainDf2.filter("metricname = 'NPV'")
                 .withColumnRenamed("my_attr","df2attr")
                 .withColumnRenamed("value","df2val")
                 .coalesce(1)
// df3、df4照猫画虎就行

这样后续Join的时候，就不会继承原始DF的海量分区了。

2. 用广播Join彻底避免shuffle #

你的每个DF只有4行，完全符合广播Join的场景——把小DF直接分发到所有Executor，根本不需要进行shuffle操作，能大幅减少集群的调度压力：

import org.apache.spark.sql.functions.broadcast

var resultDf = df1.as("dft").join(broadcast(df2), joinCols, "inner")
                 .select("dft.*","df2attr","df2val")
// 后续Join继续用广播，而且每次Join后可以再合并一次分区确保安全
resultDf = resultDf.as("dfi").join(broadcast(df3), joinCols, "inner")
                 .select("dfi.*","df3attr","df3val")
                 .coalesce(1)

3. 检查并调整Spark的shuffle分区配置 #

Spark默认的spark.sql.shuffle.partitions是200，但你的场景下数据量极小，这个值完全可以调得更低（比如1-10）。你可以在代码开头或者提交任务时设置：

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

或者提交任务时加参数：--conf spark.sql.shuffle.partitions=1

4. 排查原始DF的分区来源 #

你可以先看看原始的mainDf1/mainDf2有多少分区：

println(s"mainDf1分区数: ${mainDf1.rdd.partitions.size}")

如果原始DF的分区数本来就有几万，那过滤后每个分区可能是空的或者只有1行，但分区数还是保留着。这种情况下，除了过滤后合并分区，也可以考虑在加载原始数据时就调整分区数（比如读取文件时指定minPartitions）。

5. 简化Join逻辑，减少分步操作的开销 #

既然所有DF都是基于相同的过滤条件（metricname = 'NPV'），你可以先把所有需要的列提前准备好，然后一次性完成所有Join，减少中间步骤的分区开销：

// 先把每个DF需要的列单独提出来，过滤+重命名
val df1 = mainDf1.filter("metricname = 'NPV'")
                 .withColumnRenamed("my_attr","df1attr")
                 .withColumnRenamed("value","df1val")
                 .coalesce(1)
val df2 = mainDf2.filter("metricname = 'NPV'")
                 .select(joinCols ++ Seq("my_attr","value"): _*)
                 .withColumnRenamed("my_attr","df2attr")
                 .withColumnRenamed("value","df2val")
                 .coalesce(1)
// 一次性Join所有小DF
val resultDf = df1.join(broadcast(df2), joinCols, "inner")
                 .join(broadcast(df3), joinCols, "inner")
                 .join(broadcast(df4), joinCols, "inner")

按上面的方法调整后，分区数应该能控制在个位数，任务跑起来肯定快很多，也不会再出现内存错误了。

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