窗口函数会暴露数据倾斜，加盐需用concat+rand确保随机性，组合分区利用时间维度限流，两阶段聚合避免单点瓶颈，需调优buffer阈值防spill。
窗口函数本身不解决倾斜，反而会把倾斜问题暴露得更彻底——PARTITION BY 字段一旦分布不均，所有同 key 的行就被拉到一个 task，没有 shuffle 后的 merge 阶段缓冲，直接卡死。
为什么加盐必须用 concat + rand 而不是 hash
Spark 的 hash 函数（如 hash(user_id)）是确定性的，同一个 user_id 每次都映射到固定 salt 值，无法打散热点；而 concat(col("user_id"), lit("_"), (rand() * 10).cast("int")) 每次执行生成不同后缀，确保单个大 key 被随机分到多个子分区。但要注意：rand() 必须在 select 或 withColumn 中调用，不能放在 partitionBy 里直接用——否则会被优化器提前折叠，失去随机性。
- 错误写法：
Window.partitionBy(hash("user_id") % 10)→ 所有 500 万次 user_id=123 都落到同一子分区 - 正确写法：
df.withColumn("salted_id", concat(col("user_id"), lit("_"), (rand() * 100).cast("int"))).withColumn("rn", row_number().over(Window.partitionBy("salted_id").orderBy("ts"))) - 盐值数量建议 10–100：太少打散不足，太多导致小 key 的 group by 开销反超
组合分区比单纯加盐更可控
当业务天然带有时序属性（如日志、订单、事件），直接把时间维度引入 PARTITION BY 是最轻量的倾斜缓解方式。比如 partitionBy("user_id", "date_trunc('day', ts)"))，能把单日 500 万次的用户行为，按天切分成独立分区，既避免全量打散的计算开销，又天然限流——头部用户再活跃，单日数据量也有物理上限。
- 适用前提：时间字段稳定、非空、粒度合理（用
'hour'可能产生过多分区，'month'又可能不够细） - 注意
date_trunc返回的是 timestamp 类型，和原始ts类型一致，不会引发隐式转换 - 如果原始表已按
user_id和日期做了 bucket 或 partition，组合分区还能复用底层存储布局，减少 shuffle
两阶段聚合：先拆再收，绕过单点瓶颈
加盐或组合分区后，你拿到的是“每个子组的局部结果”，比如每个 user_id + salt 分区里的最早点击时间。但最终要的是“每个 user_id 的全局最早时间”——这时不能直接 GROUP BY user_id，否则又回到原点。正确做法是第二阶段用窗口函数对子组结果再做一次 row_number() OVER (PARTITION BY user_id ORDER BY local_min_ts)，取 rn = 1 的记录。
- 第一阶段输出必须包含原始 key（如
user_id）和局部指标（如min_ts），且不能丢弃user_id字段 - 第二阶段的
Window.partitionBy("user_id")数据量已大幅下降（从 500 万行 → 几百行），不再构成瓶颈 - 整个流程本质是把“单点重计算”转为“多点轻计算 + 单点轻合并”，executor 内存压力下降 90%+
真正容易被忽略的是：窗口函数的 buffer 行为。默认 spark.sql.windowExec.buffer.in.memory.threshold=4096，一旦单个分区超过该行数就触发磁盘 spill；而倾斜分区动辄几十万行，频繁读写本地磁盘会拖垮整个 stage。上线前务必检查该参数是否适配你的最大分区规模——不是越大越好，而是要略高于 P99 分区大小，避免无谓 spill，又不至于吃光 executor 内存。
