窗口函数不缓解倾斜反而暴露倾斜,PARTITION BY user_id因长尾导致单task OOM;加盐(如concat(user_id, "_", (rand()*10).cast("int")))打散热点,再两阶段聚合(先salt分组,后按原user_id窗口取代表)才是生产级方案。

窗口函数本身不缓解倾斜,反而会把倾斜问题暴露得更彻底——PARTITION BY 字段一旦存在长尾分布,对应 task 就会卡死。必须主动拆解热点分区,不能指望 Spark 自动处理。
为什么 PARTITION BY user_id 会触发 OOM?
Spark 窗口函数在执行时,PARTITION BY 字段决定数据 shuffle 到哪个 task;如果某个 user_id 出现 500 万次,所有相关行都会落到同一个 task,没有并行度可言。执行计划里出现 Exchange SinglePartition 或 WholeStageCodegen 下挂超长耗时 task,就是明确信号。
- 不是数据量太大,而是单个 key 的数据太集中
-
Window.orderBy(...)(没写PARTITION BY)等于全表一个分区,绝对禁止 - 即使只想要最新一条记录,也必须先按去重键分组,
row_number()才有意义
加盐(salting)是最常用且低侵入的解法
核心是把大分区打散成多个子分区,让计算分散到不同 task,最后再合并结果。关键不是“加随机值”,而是控制盐值数量和分布。
- 盐值数量建议 10–100:太少起不到打散效果,太多会增加小 key 的 group by 开销
- 推荐写法:
concat(col("user_id"), lit("_"), (rand() * 10).cast("int")) - 加盐后仍需保留原始
user_id字段,用于第二阶段聚合或 join - 注意:盐值列不能参与业务逻辑判断,仅作 shuffle 分散用
两阶段聚合 + 窗口函数才是生产级方案
单靠加盐只能缓解,真正需要精确结果(如每个用户首次点击时间、最新订单状态)时,必须拆成两个 stage:第一阶段局部聚合,第二阶段全局归并。
- 第一阶段:
groupBy("salted_user_id").agg(min("event_time").alias("min_ts"), ...) - 第二阶段:用
row_number() over (partitionBy("user_id") order by "min_ts")取每组代表 - 避免在
PARTITION BY字段上直接用collect_set(),一个分区收集几百万元素会拖垮 executor - 如果只是近似去重计数,优先用
approx_count_distinct(x, 0.01),比窗口内精确计数快 5–10 倍
最容易被忽略的是:缓存输入 DataFrame。如果同一张表被多次用于不同窗口计算,不 cache() 会导致重复 shuffle 和全量重算——这点在调试阶段常被跳过,但上线后就是性能黑洞。

















