Spark：如何拆分单个Executor上的海量数据？解决sampleBy后的分区问题

问题根源拆解 #

你遇到的核心问题是sampleBy处理极端不平衡数据时引发的分区倾斜：当目标类别分布差距极大，sampleBy为了实现均匀采样，会对少数类别抽取极高比例的样本，导致这些样本集中到单个/极少数分区里。后续单纯用df.repartition(n)依赖默认哈希分区，没法打破这种集中——因为哈希键（默认是全列哈希）依然高度重合，数据还是会扎堆到同一个Executor。

可行解决方案 #

1. 采样后加盐打散分区 #

不要直接用repartition，给目标类别字段加一个随机盐值，把盐和类别字段一起作为分区键，强制打散同类别数据：

// 假设你的目标类别字段是target_col，n是期望的分区数
import org.apache.spark.sql.functions.{rand, col}
import org.apache.spark.sql.types.IntegerType

val saltedSampledDf = sampledDf.withColumn("salt", (rand() * n).cast(IntegerType))
val balancedDf = saltedSampledDf.repartition(n, col("target_col"), col("salt"))

盐值会把同一类别的数据均匀分配到n个分区里，彻底解决单Executor负载集中的问题。

2. 替换sampleBy：手动分层采样 #

如果sampleBy本身是倾斜的源头，换成分层处理的方式：大类别按比例采样并提前打散，小类别全量保留，最后合并再分区：

// 拆分出占比极高的大类别和其他小类别
val majorClassDf = originalDf.filter(col("target_col") === "高占比类别值")
val minorClassesDf = originalDf.filter(col("target_col") =!= "高占比类别值")

// 大类别采样后先打散，小类别直接保留，最后合并再统一分区
val sampledMajorDf = majorClassDf.sample(0.02).repartition(n/2) // 采样比例按需调整
val finalBalancedDf = sampledMajorDf.union(minorClassesDf).repartition(n)

这种方式从采样阶段就避免了单类别数据过度集中，从根源减少倾斜。

3. 配套调整Spark Shuffle参数 #

在执行分区操作前，先设置匹配的Shuffle参数，让集群资源适配分区需求：

// 设置Shuffle分区数和执行器核心数（根据你的集群资源调整）
spark.conf.set("spark.sql.shuffle.partitions", n)
spark.conf.set("spark.executor.cores", "4")

默认的spark.sql.shuffle.partitions是200，要和你期望的最终分区数一致，避免后续自动Shuffle时再次出现数据扎堆。

验证分区效果 #

做完分区后，用下面的代码检查每个分区的数据量分布：

balancedDf.rdd.glom().map(_.length).collect()

如果返回的数值分布均匀，说明分区拆分成功，Shuffle读的负载也会自动分散到各个Executor上。

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