怎样在Spark SQL中避免窗口函数引发的全局单点Shuffle-SQL-PHP中文网

不写 partition by 会触发单点 shuffle，导致 oom 和任务卡死；必须用高基数字段分区，避免倾斜，并合并多 rank 表达式以减少 shuffle 次数。

怎样在spark sql中避免窗口函数引发的全局单点shuffle

不写 PARTITION BY 就等于主动触发单点 Shuffle

Spark SQL 里只要窗口函数用了 OVER 却没写 PARTITION BY，就会强制把所有数据 shuffle 到单个分区。日志里会出现 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition —— 这不是提示，是明确警告你正在制造瓶颈。10GB 数据可能直接 OOM，任务卡死在 Stage 0，executor 内存打满。

错误写法示例：ROW_NUMBER() OVER (ORDER BY ts DESC)；正确写法必须带分区：ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY ts DESC)。

常见误判是以为“只要写了 ORDER BY 就够了”，但 ORDER BY 只负责排序逻辑，PARTITION BY 才决定数据怎么分组、是否并行。业务上真正需要“全表排第几”的场景极少，绝大多数排名、累计求和、移动平均都有天然分组维度（如 user_id、order_id、event_time::date）。

选错 PARTITION BY 字段比不选还危险

用低基数或倾斜字段做分区键，会把单点问题变成局部单点问题。比如按 country 分区算销量排名，而中国数据占 95%，那中国分区仍会卡住整个 stage。

选字段时注意以下几点：

优先选高基数、分布均匀的字段：如 user_id、order_id、event_time::date（注意去重后基数）
避免低基数字段：如 status（只有 'active'/'inactive'）、is_paid（布尔值）
如果必须用倾斜字段，先用 WHERE 过滤掉占比过大的值，或对热点值单独加盐处理（如 CONCAT(country, '_', FLOOR(RAND() * 10))）
验证分布：运行 SELECT country, COUNT(*) FROM t GROUP BY country ORDER BY 2 DESC LIMIT 5，看 Top 5 是否超过总量 50%

多个 rank() 表达式别拆开写

想给多列同时加排名，别写多个 withColumn("rank_a", rank().over(...)) 链式调用——每次都会触发新 job，物理计划无法合并，shuffle 次数翻倍。

Mootion

Mootion是一款将文本、脚本、图片和想法生成视频与 3D 动画的 AI 创作平台。

下载

正确做法是用 selectExpr 一次性声明全部表达式，Catalyst 能识别它们共享同一套 PARTITION BY + ORDER BY 逻辑，只 shuffle 一次：

df.selectExpr(
  "rank() OVER (PARTITION BY user_id ORDER BY ts) AS rank_ts",
  "rank() OVER (PARTITION BY user_id ORDER BY amount DESC) AS rank_amt"
)

如果各列排序逻辑不同（比如一列升序、一列降序），确实无法合并，那就接受多次 shuffle，但务必确保每次都有 PARTITION BY，不能退化为全局排序。

分区倾斜时，加盐不是万能解药

当 PARTITION BY user_id 遇到头部用户（某 user_id 出现 500 万次），Spark 不会自动拆分这个分区。加盐（如 CONCAT(user_id, '_', (rand() * 10).cast("int"))）能把大分区打散，但盐值数量要合理（通常 10–100），太多反而增加小 key 的 group by 开销。

更稳妥的替代方案包括：