怎么在Spark SQL中针对倾斜Key进行两阶段分组聚合？-SQL-PHP中文网

倾斜key是指出现频率极高、导致shuffle时大量数据集中到同一partition的key（如""、"null"、热点商品id）；两阶段聚合通过第一阶段为倾斜key加随机前缀并局部聚合、第二阶段去前缀合并，分散计算压力，从而缓解数据倾斜。

怎么在spark sql中针对倾斜key进行两阶段分组聚合？

什么是倾斜Key，为什么两阶段聚合能缓解它

Spark SQL中遇到groupBy或agg后任务长时间卡在99%、个别task运行时间远超其他task，大概率是数据倾斜。典型表现是Shuffle Write量巨大，且某些reducer接收的数据量是其他reducer的几十甚至上百倍。根本原因是某些key（比如“”、"null"、用户ID为"000000"或热点商品ID）出现频率极高，导致shuffle阶段大量数据涌向同一个partition。

两阶段聚合的核心思路是：先局部打散热点key，加随机前缀做预聚合，再全局去前缀合并。它不依赖业务改写逻辑，也不强制要求加盐（salt）字段，而是用SQL原生能力实现可控的二次分组。

第一阶段：对key加随机前缀并局部聚合

这一步要让原本相同的热点key被分散到多个partition中，同时保留足够信息用于后续还原。关键点是用rand()生成随机数，但不能直接concat(key, '_', rand())——因为rand()在单行内多次调用会返回不同值，导致同一行数据在后续无法对齐。

正确做法是先用rand()生成一个稳定的前缀，再拼接：

SELECT
  CASE WHEN key IN ('', 'null', '000000') THEN concat(cast(floor(rand() * 100) as string), '_', key)
       ELSE key
  END AS skewed_key,
  sum(value) AS local_sum,
  count(*) AS local_count
FROM your_table
GROUP BY skewed_key

floor(rand() * 100)控制打散粒度，100表示最多拆成100份；太小（如10）可能仍不够，太大（如1000）会增加小task数量和调度开销
必须用CASE WHEN只对已知倾斜key加前缀，否则所有key都打散，反而降低非倾斜key的聚合效率
不要用md5(key || rand())之类不可逆方式，后续无法还原原始key

第二阶段：去掉前缀，合并同一原始key的所有局部结果

上一阶段输出的skewed_key形如"42000000"或"77""，需要提取原始key并再次聚合：

SELECT
  CASE WHEN skewed_key RLIKE '^\d+_(.*)$' THEN regexp_extract(skewed_key, '^\d+_(.*)$', 1)
       ELSE skewed_key
  END AS key,
  sum(local_sum) AS total_sum,
  sum(local_count) AS total_count
FROM (
  -- 第一阶段SQL嵌套在此
) t
GROUP BY key

regexp_extract比split(skewedkey, '', 2)[1]更安全，避免空字符串切分出界
如果原始key本身含下划线（如"userabc"），就不能用做分隔符，得换成不会出现在key中的字符，比如"u0001"（ASCII 1），对应写成concat(cast(floor(rand() * 100) as string), 'u0001', key)
注意RLIKE在某些Spark版本（如3.0以下）对空字符串匹配可能异常，可改用length(skewed_key) > 3 AND substr(skewedkey, 3, 1) = ''等更保守判断