Spark技术问题：无需聚合时如何加盐并将数据均匀分配至执行器

问题分析 #

你遇到的核心问题是：尽管添加了salt并重新分区，但数据仍集中在单个执行器，这大概率是以下两个原因之一：

1. salt生成方式导致分布不均：rand()生成的随机数在Spark分布式环境中可能存在局部不均匀性，当数据集规模极大时，部分分区的salt值可能高度集中。
2. 后续操作触发二次shuffle：如果写入磁盘时使用了partitionBy("y")，Spark会自动按y重新分区，完全抵消之前加盐repartition的效果——倾斜列y的某个取值占比过高，所有该取值的数据会被shuffle到同一节点写入，导致执行器负载集中。

解决方案 #

1. 换用确定性的salt生成方式 #

放弃rand()，改用全局唯一ID取模生成salt，确保每个分区的行数绝对均匀：

from pyspark.sql import functions as F
from pyspark.sql.types import IntegerType

n = 2000
# 用全局递增ID生成salt，避免rand()的随机波动
train = train.withColumn("salt", (F.monotonically_increasing_id() % n).cast(IntegerType()))
# 仅以salt作为分区键，确保数据均匀分布
train = train.repartition(n, "salt")

如果数据集本身有业务唯一ID（如用户ID、订单ID），也可以用该ID取模：

train = train.withColumn("salt", (F.hash("business_id") % n).cast(IntegerType()))

2. 避免写入时触发按y的shuffle #

• 如果不需要按y分区存储：直接执行写入操作，不要添加partitionBy("y")，此时数据会按salt的分区均匀写入磁盘。
• 如果必须按y分区存储：将salt也加入分区列，让同一个y值被拆分为n个子分区，分散写入压力：

train.write.partitionBy("y", "salt").parquet("/path/to/output")

这样每个y值下会生成n个salt子目录，每个子目录对应的数据量约为该y值总行数的1/n，执行器负载会被均匀分摊。

3. 验证salt分布 #

在repartition后先统计salt的行数分布，确认是否均匀：

train.groupBy("salt").count().orderBy(F.desc("count")).show(10)

如果统计结果中各salt的行数差异极小，说明分区已经均匀，后续写入时就不会出现单执行器集中数据的问题。

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