GROUP BY做统计，一个Task卡死跑不完——COUNT(DISTINCT)为什么特别容易倾斜？

-- 场景A：COUNT(*)——可以在Map端局部预聚合
SELECT event_date, COUNT(*) AS pv
FROM user_events
GROUP BY event_date;
-- Spark内部执行：先在每个分区内COUNT(*) → 再按event_date Shuffle → 最终SUM
-- 如果event_date='2024-01-01'有1亿行，Map端先聚合成1行（count=100000000），
-- Shuffle的数据量非常小，不会倾斜
 
-- 场景B：COUNT(DISTINCT)——无法Map端预聚合
SELECT event_date, COUNT(DISTINCT user_id) AS dau
FROM user_events
GROUP BY event_date;
-- Spark内部执行：必须把所有数据Shuffle到按event_date分区的Task
-- event_date='2024-01-01'的1亿行必须全部发到同一个Task
-- 这个Task要在内存里维护一个去重集合，数据量大=必然慢甚至OOM

-- 原始查询：按渠道统计广告曝光量，某个渠道数据量极大
SELECT channel_id, SUM(exposure) AS total_exposure
FROM ad_events
GROUP BY channel_id;
 
-- 两阶段聚合：第一阶段加随机前缀打散，第二阶段去掉前缀汇总
SELECT
  channel_id,
  SUM(partial_sum) AS total_exposure  -- 第二阶段：按真实key汇总
FROM (
  SELECT
    channel_id,
    CONCAT(CAST(FLOOR(RAND() * 10) AS STRING), '_', channel_id) AS salted_key,
    SUM(exposure) AS partial_sum  -- 第一阶段：加前缀后局部聚合
  FROM ad_events
  GROUP BY channel_id, salted_key  -- 用salted_key分散数据到10个Task
) t
GROUP BY channel_id;
 
-- RAND()*10意味着同一个channel_id的数据被打散到10个Task处理
-- 每个Task处理1/10的数据，再汇总10个局部结果

-- 原始有倾斜的写法
SELECT event_date, COUNT(DISTINCT user_id) AS dau
FROM user_events
GROUP BY event_date;
 
-- 方案一：拆成子查询（推荐，精确计算）
SELECT event_date, COUNT(user_id) AS dau
FROM (
  -- 内层：先按(event_date, user_id)去重，把GROUP BY换成两个维度
  SELECT event_date, user_id
  FROM user_events
  GROUP BY event_date, user_id  -- 可以加随机前缀进一步打散
) t
GROUP BY event_date;
 
-- 效果分析：
-- 内层：按(event_date, user_id)分组，每组只有少量数据，天然打散
-- 外层：COUNT(user_id)是普通COUNT，可以Map端预聚合，不会倾斜

-- 方案二：approx_count_distinct（近似计算，误差1-2%，但极快）
SELECT event_date, approx_count_distinct(user_id) AS dau_approx
FROM user_events
GROUP BY event_date;
 
-- Spark对应的函数：approx_count_distinct(col, rsd)
-- HiveQL对应的函数：ndv(col)  -- HyperLogLog算法
-- rsd = relative standard deviation，误差率，默认0.05（5%误差）
-- 如果业务允许误差，这是最快的方案，避免了Shuffle
 
-- 什么时候用近似计算？
-- 趋势分析（看DAU趋势，1-2%误差可接受）
-- 实时监控大盘（精度要求低，速度要求高）
-- 注意：财务类、结算类绝对不能用近似计算

两种方案的对比选择

方案	精确度	性能	适用场景	我的选择
子查询拆分	精确	好（避免了大Shuffle）	业务报表、精确指标	首选
approx_count_distinct	约95-99%	极快（无Shuffle）	实时大盘、趋势分析	速度优先时选
两阶段打散COUNT(DISTINCT)	❌ 会出错	-	不能用	绝对不用

第三行特意列出来，因为面试中有人会说「给user_id加随机前缀打散」——这是错的！加了前缀后，同一个user_id会被计为多个不同的用户，COUNT(DISTINCT)结果偏大。要去重，必须保留原始的user_id值。

面试高频追问

追问一：子查询拆分方案里，内层GROUP BY还是可能倾斜怎么办？

内层GROUP BY (event_date, user_id)的key是复合key，理论上比单独的event_date粒度更细，倾斜概率大幅降低。但如果event_date维度本身就很倾斜（比如某一天有10亿行），还可以再加一层打散：在event_date之外，再加一个随机前缀（1-50）形成(event_date, prefix, user_id)的三级key。先局部去重，再按(event_date, user_id)去重，最后COUNT。实际生产中两层通常够用。

追问二：Spark AQE能自动处理GROUP BY倾斜吗？

Spark 3.0之后的AQE（Adaptive Query Execution）有自动处理Shuffle数据倾斜的能力，叫做Skew Join Optimization，但它主要针对的是JOIN倾斜，而不是GROUP BY聚合倾斜。对于GROUP BY COUNT(DISTINCT)，AQE目前还不能自动处理，需要手动改写SQL。开启AQE的方法：spark.sql.adaptive.enabled=true，spark.sql.adaptive.skewJoin.enabled=true。

追问三：在Hive里和在Spark SQL里，COUNT(DISTINCT)的行为有什么不同？

Hive有一个专门的参数 hive.groupby.skewindata=true，开启后会自动做两阶段GROUP BY聚合。但这个参数对COUNT(DISTINCT)同样无效——Hive也没办法自动处理COUNT(DISTINCT)的倾斜，还是得手动改写。Spark SQL没有这个参数，但AQE提供了更智能的自适应处理。实际上不论哪个引擎，COUNT(DISTINCT)的倾斜都需要工程师手动识别和改写。

实战踩坑

我在某短剧出海公司做广告效果分析时，有一个每日计算广告创意UV（按广告素材ID去重计算曝光用户数）的作业，跑了快2个小时。一看Spark UI，有一个Task的Shuffle Read数据量是其他Task的800倍——某一个爆款广告素材的曝光量占了当天总量的40%，所有数据集中到一个Task里做COUNT(DISTINCT)。

-- 原始有问题的SQL
SELECT
  creative_id,
  COUNT(DISTINCT user_id) AS uv,
  SUM(exposure) AS pv
FROM ad_exposure
WHERE dt = '2024-08-01'
GROUP BY creative_id;
 
-- 改写后：拆分COUNT(DISTINCT为子查询
SELECT
  t1.creative_id,
  t2.uv,
  t1.pv
FROM (
  -- PV不受影响，正常GROUP BY
  SELECT creative_id, SUM(exposure) AS pv
  FROM ad_exposure
  WHERE dt = '2024-08-01'
  GROUP BY creative_id
) t1
JOIN (
  -- UV单独拆成子查询去重后COUNT
  SELECT creative_id, COUNT(user_id) AS uv
  FROM (
    SELECT creative_id, user_id
    FROM ad_exposure
    WHERE dt = '2024-08-01'
    GROUP BY creative_id, user_id  -- 先去重
  ) dedup
  GROUP BY creative_id
) t2 ON t1.creative_id = t2.creative_id;
 
-- 改写后：从1小时58分钟降到11分钟

注意：这里把PV和UV分开两个子查询计算，是因为混在一起写COUNT(DISTINCT)和SUM，Spark会用同一套执行计划，没法分别优化。拆开来，PV子查询走正常聚合，UV子查询走去重+COUNT，各自独立优化。

总结

GROUP BY倾斜靠两阶段聚合打散，核心是Map端先局部聚合减少Shuffle数据量。COUNT(DISTINCT)不能直接打散，正确做法是拆成子查询先去重再COUNT，或者业务允许误差时用approx_count_distinct。面试时能说清楚「为什么COUNT(DISTINCT)不能直接打散」以及「正确的改写方式」，就能和大多数候选人拉开差距。

下一篇聊JOIN倾斜——两张大表JOIN时某个key的数据量超大，该怎么处理？Broadcast Join、加盐打散、拆分分别适合什么场景？

你们生产环境有没有遇到过COUNT(DISTINCT)倾斜？是怎么解决的？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长