加盐+两层聚合是解决COUNT(DISTINCT)数据倾斜最稳方案：先用hash(user_id,item_id)%N加盐打散，再按(user_id,salt)去重后汇总，避免单点过载。
数据倾斜导致COUNT(DISTINCT)触发大量Shuffle
当分组键（如 user_id）分布极不均匀时，COUNT(DISTINCT) 的底层实现必须把所有相同键的记录拉到同一个计算节点做去重——这会引发严重的数据倾斜。比如 99% 的点击日志都来自 10 个超级用户，那这 10 个 user_id 对应的中间数据就会全部堆积在一个 Reduce 或 Executor 上，其他节点空转。
常见错误现象包括：
- 任务卡在 99% 进度不动，
Shuffle Write高达几十 GB，而其他任务早已完成 - Executor OOM 报错：
java.lang.OutOfMemoryError: Java heap space - Spark UI 中看到某 task 执行时间是其他 task 的百倍以上
Hive/Spark 默认执行策略加剧倾斜
默认情况下，COUNT(DISTINCT) 会走两阶段聚合：第一阶段按分组键局部去重（partial_merge），第二阶段全局合并。但若倾斜发生在第一阶段，局部去重后仍残留巨量唯一值（例如一个 user_id 对应上百万个不同 item_id），第二阶段就无法缓解压力。
关键参数差异：
- Hive 中
hive.map.aggr=true可开启 Map 端预聚合，但对严重倾斜无效 - Spark SQL 中
spark.sql.adaptive.enabled=true能自动拆分倾斜分区，但仅限 AQE 启用且版本 ≥ 3.2 - 两者都不支持对
DISTINCT内部值做盐值（salting）干预，必须手动改写
手动改写：加盐 + 两层聚合是最稳解法
核心思路是把原本集中在一个 key 上的 distinct 值，打散到多个子 key，再合并。适用于 Hive、Spark、Flink SQL 等几乎所有引擎。
以 Spark SQL 为例：
SELECT user_id, COUNT(DISTINCT item_id) AS cnt FROM events GROUP BY user_id;
→ 改写为：
-- 第一层：加盐打散 SELECT user_id, floor(rand() * 10) AS salt, -- 生成 0~9 的随机盐值 item_id FROM events <p>-- 第二层：先按 (user_id, salt) 去重，再按 user_id 汇总 SELECT user_id, COUNT(DISTINCT concat(item<em>id, '</em>', salt)) AS cnt FROM ( SELECT user_id, floor(rand() * 10) AS salt, item_id FROM events ) t1 GROUP BY user_id; </p>
注意点：
- 盐值数量（如
* 10）要略大于预期倾斜 key 的数量，太少仍会倾斜，太多增加 shuffle 量 -
concat(item_id, '_', salt)确保不同 salt 下的相同item_id不被误去重 - 如果
item_id本身含下划线，改用不可见字符如concat(item_id, '\u0001', salt)
别忽略 NULL 和重复采样带来的误差
加盐方案虽有效，但有两个容易被跳过的坑：
-
rand()在 Spark 中是非确定性函数，同一行数据在不同 stage 可能生成不同 salt，导致漏计或重复计 —— 必须用hash(user_id, item_id) % 10替代rand() - 原始数据中
item_id为NULL时，COUNT(DISTINCT)自动忽略；但加盐后concat(NULL, '_0')是NULL，仍会被忽略，逻辑一致；可放心 - 如果业务允许少量误差，优先考虑
APPROX_COUNT_DISTINCT(item_id)，它基于 HyperLogLog，内存开销固定且无倾斜风险
