结论：应避免单 SELECT 中多个 COUNT(DISTINCT)，因其触发 Expand 膨胀导致 Shuffle 暴增；单个 COUNT(DISTINCT) 易 OOM 是因 HashAggregate 全量加载去重字段至内存；须拆为独立子查询 JOIN 或用随机前缀两阶段预聚合，并确保分区裁剪、AQE 正确启用。
直接说结论：别在一个 SELECT 里写多个 COUNT(DISTINCT)，也别指望 Spark 自动帮你扛住 UV 级去重——它会炸内存、卡 Shuffle、甚至 OOM 报错退出。
为什么单个 COUNT(DISTINCT) 就容易 OOM？
Spark SQL 默认用 HashAggregate 实现 COUNT(DISTINCT)，本质是把所有待去重值（比如 uid）全 load 进内存建哈希表。1 亿个 UUID，每个占 36 字节，光 key 就要 3.6 GB；再加上哈希桶、扩容、JVM 对象头开销，很容易突破 executor 的 spark.executor.memory 或 spark.sql.adaptive.enabled 默认限制。
- 错误现象：
Container killed by YARN for exceeding memory limits或java.lang.OutOfMemoryError: Java heap space - 真实瓶颈不在磁盘或网络，而在单个 task 的堆内存 —— 即使你开了 20 个 core，只要一个 reducer 拿到倾斜的
uid分区，就挂 -
spark.sql.adaptive.enabled=true在 3.2+ 能自动拆分倾斜 partition，但前提是数据有明显 key 倾斜且你没关掉 AQE；默认不开启，或版本低于 3.2 时完全无效
多个 COUNT(DISTINCT) 必须拆成子查询 JOIN
写成 SELECT dt, COUNT(DISTINCT uid), COUNT(DISTINCT aid), COUNT(DISTINCT pid) 这种形式，Spark 会插入 Expand 节点：1 行原始数据 → 复制成 N 行（N = 去重字段数），每行只标记一个字段参与去重。30 个字段？就是 1 行 → 30 行，Shuffle 数据量翻 30 倍。
- 必须改写为独立子查询 +
JOIN，例如： SELECT t1.dt, t1.uid_cnt, t2.aid_cnt, t3.pid_cnt FROM (SELECT dt, COUNT(DISTINCT uid) AS uid_cnt FROM events WHERE dt = '2026-10-14' GROUP BY dt) t1 JOIN (SELECT dt, COUNT(DISTINCT aid) AS aid_cnt FROM events WHERE dt = '2026-10-14' GROUP BY dt) t2 ON t1.dt = t2.dt JOIN (SELECT dt, COUNT(DISTINCT pid) AS pid_cnt FROM events WHERE dt = '2026-10-14' GROUP BY dt) t3 ON t1.dt = t3.dt
- 两个硬性要求：
WHERE条件必须完全一致（否则分组键对不上）、每个子查询结果集不能太大（理想几百～几千行） - 如果底层是 Hive 表，确保
dt是分区字段，且子查询中显式写了WHERE dt = '2026-10-14'，否则分区裁剪失效，扫全表
单字段 COUNT(DISTINCT) 倾斜时加随机前缀预聚合
当某个 visit_type 下的 pv_id 特别多（比如首页曝光占全站 70%），直接 COUNT(DISTINCT pv_id) 会导致 reducer 热点。这时不能靠增加 parallelism 解决，得改数据分布逻辑。
- 标准两阶段法：
- 第一层：按
visit_type+hash(pv_id) % 100分组，每个 group 内去重计数 - 第二层：按
visit_type汇总各前缀组的计数和 - 示例 SQL：
SELECT visit_type, SUM(cnt) AS pv_cnt FROM ( SELECT visit_type, COUNT(DISTINCT pv_id) AS cnt FROM events WHERE dt = '2026-10-14' GROUP BY visit_type, CAST(rand() * 100 AS INT) ) t GROUP BY visit_type
- 注意：
rand()要放在GROUP BY里，不能放外层，否则无法打散；CAST(... AS INT)避免小数导致分组失败
别忽略 Catalyst 和 AQE 的实际开关状态
很多人开了 spark.sql.adaptive.enabled=true 就以为万事大吉，但真正起效需要配套条件：
-
spark.sql.adaptive.skewJoin.enabled=true：只对JOIN倾斜生效，对GROUP BY无效 -
spark.sql.adaptive.localShuffleReader.enabled=true：减少本地磁盘 shuffle 文件读取压力，但需配合spark.sql.adaptive.coalescePartitions.enabled=true - 最易被忽略的一点：
EXPLAIN FORMATTED查看物理计划，确认是否真出现了AdaptiveSparkPlan节点 —— 如果没有，说明 AQE 根本没触发，可能是数据量未达阈值、或用了不支持的算子（如collect_list） - 临时验证方式：在 SQL 前加
/*+ COALESCE(200) */hint 强制调整 reduce task 数，比调参更快定位是不是并行度问题
真正卡住性能的往往不是语法写法，而是分区裁剪漏了、索引没建、AQE 实际没生效、或者把 10 个 COUNT(DISTINCT) 塞进一行还指望 Spark 自动优化 —— 这些地方一错，再大的集群也救不回来。
