如何在Spark SQL中处理倾斜键的分组聚合性能瓶颈-SQL-PHP中文网

group by卡在个别task上是因为shuffle阶段倾斜key（如占40%数据的sku_1000001）被集中到单个task处理，导致负载不均和oom；加盐需两阶段：先对倾斜key局部打散（如concat(key,'_',rand*500)），再全局聚合，且盐值数量应为shuffle分区数的1/3～1/2。

如何在spark sql中处理倾斜键的分组聚合性能瓶颈

为什么`GROUP BY`会卡在个别Task上？

因为Shuffle阶段必须把相同key的数据拉到同一个Task处理。如果某个key（比如商品ID 'SKU_1000001'）占了全量数据的40%，它就会被塞进单个Task，而其他999个key分散在其余Task里——结果就是99%的Task 2秒跑完，1个Task跑2小时，还可能OOM。

`GROUP BY`加盐（salting）怎么写才不翻车？

核心是两阶段：先局部打散再全局合并。不能直接用rand()，得控制盐值范围，否则拆太碎反而增加Shuffle开销。

盐值数量建议设为当前spark.sql.shuffle.partitions的1/3～1/2（比如分区数2000，盐值就取500～800）
倾斜key必须显式识别出来，靠SELECT key, COUNT(*) c FROM tbl GROUP BY key ORDER BY c DESC LIMIT 10查前10大key
盐值只加给倾斜key，正常key保持原样，避免无谓膨胀

示例SQL（假设倾斜key是'SKU_1000001'）：

SELECT key, SUM(cnt) AS total_cnt
FROM (
  SELECT 
    CASE WHEN key = 'SKU_1000001' 
      THEN CONCAT(key, '_', CAST(FLOOR(RAND() * 500) AS STRING)) 
      ELSE key 
    END AS key,
    COUNT(*) AS cnt
  FROM tbl
  GROUP BY 
    CASE WHEN key = 'SKU_1000001' 
      THEN CONCAT(key, '_', CAST(FLOOR(RAND() * 500) AS STRING)) 
      ELSE key 
    END
) t
GROUP BY key

用`spark.sql.adaptive.skewedPartitionMaxSplits`能省事吗？

能，但只适用于Spark 3.2+且开启自适应查询执行（AQE）的场景。它会在运行时自动检测倾斜分区并切分，不用改SQL。不过有硬限制：

柒源写作

柒源写作是一款面向论文、报告和公文材料的中文 AI 写作平台。

下载

默认值是5，最大只能设到10，对极端倾斜（比如单key占90%）效果有限
必须同时开启spark.sql.adaptive.enabled=true和spark.sql.adaptive.localShuffleReader.enabled=true
不适用于所有算子，目前主要覆盖JOIN和AGG，UNION或WINDOW函数不生效
如果你的任务已经因OOM失败，AQE根本没机会触发——得先保证能跑起来

聚合前过滤空值或脏数据为什么关键？

很多倾斜其实不是业务key本身热，而是大量NULL、空字符串、默认值（如'UNKNOWN'）被当成了合法key参与分组。这些“伪倾斜”完全没必要加盐。

先执行SELECT key, COUNT(*) FROM tbl WHERE key IS NULL OR key = '' GROUP BY key确认占比
在主SQL里用WHERE key IS NOT NULL AND TRIM(key) != ''提前过滤
如果业务允许，把NULL映射成一个固定低频key（如'KEY_NULL'），避免它们全挤在一个分区
注意：FILTER下推必须发生在Shuffle之前，所以过滤条件要写在GROUP BY上游，别放在HAVING里

真实环境里最容易被忽略的，是倾斜key的动态性——今天热的'SKU_1000001'，明天可能是'SKU_2000005'。硬编码key的加盐方案上线两周后就失效了，得配合采样逻辑或元数据监控才能长期有效。