开头我从一个真实场景说起某数据分析师拿着一个跑了几十分钟的明细查询来找我SQL 很简单核心就是SELECT COUNT(DISTINCT uid) FROM 某个几亿行的大表但任务一直卡在某个 stage磁盘溢写严重集群其它任务也受到牵连。这是典型的 Spark SQL 中 distinct 优化问题。很多人一听到 distinct 慢第一反应是加资源、调并行度但真正的问题往往不在资源而在执行计划本身。这篇文章我想把你带进一个完整排查链路先从 distinct 的物理执行原理看起再对比 group by 去重、多字段 count distinct、数据倾斜加盐、近似去重、AQE 与分桶优化最后用一个模拟项目的优化复盘把整条思路串起来。无论你是刚接触 Spark SQL 的数据分析师还是已经有一定经验的开发这篇文章都会给你一套可以直接抄作业的排查清单和优化思路。1. distinct 慢先搞清楚它到底做了什么1.1 一张图上讲明白 distinct 的两个阶段在不开启 AQE 等特殊机制的情况下SELECT DISTINCT col FROM table或者SELECT COUNT(DISTINCT col)的物理执行计划里一定会有两个核心算子Exchange和HashAggregate。Exchange就是 shuffleSpark 会根据col的哈希值把数据分发到不同的分区里去。比如分区数是 200所有 hash 值相同的col会被送到同一个分区。然后每个分区内部做HashAggregate把本分区内重复的col先去掉。等每个分区都去重完还需要Stage级的全局汇总这时候往往会出现第二次Exchange把已经局部去重的数据再汇到一个或少数几个分区做最终聚合。整个过程你可以理解为先按 key 分桶再桶内去重最后汇总。这个设计本身没问题问题是当 key 的基数很大、明细行数又特别多时shuffle 的数据量几乎等于全表数据量网络传输和磁盘溢写就成了瓶颈。我见过不少人拿着EXPLAIN输出看半天其实只需要关注几类信息有没有Exchange hashpartitioning分区数是多少HashAggregate是 partial 还是 final有没有Skew标记。只要看到有一两个 stage 处理的数据量远大于其它 stage基本就可以判断瓶颈到底在哪里了。1.2 判断瓶颈数据量巨大、内存不足、还是数据倾斜同样是 distinct 慢背后通常是三类原因处理方向完全不一样。第一类是全表数据量太大。比如几亿行明细distinct 的字段又没什么过滤性整个 shuffle 要搬的数据量可能是几十 GB。这时候单纯调大 executor 内存没用因为瓶颈在网络 IO而不是单节点计算能力重点是减少输入数据量或改变计算方式。第二类是内存压力。HashAggregate需要在内存里维护一个哈希表来记录已经见过的 key。如果单个分区内的 distinct key 基数很高哈希表过大Spark 会触发 spill把部分数据写到磁盘。这种情况在日志里能看到spill to disk指标明显上涨任务时间主要花在了磁盘 IO 上。第三类是数据倾斜。比如某个 uid 占了全量的 60%按 uid 做 hash 分区后大量数据被分到同一个分区这个分区的 reducer 就成了整个任务的短板其它分区早跑完了就等它一个。这种情况加 executor、加并行度都很难解决必须从数据分布下手。一个快速判断办法是看 stage 的 task 耗时分布正常情况下各 task 耗时比较均匀如果出现一个 task 跑 30 分钟其它 task 跑 1 分钟的形态优先怀疑倾斜。2. distinct 和 group by 去重执行计划里的同与不同2.1 单字段去重时两者几乎等价很多老手会告诉你用 group by 代替 distinct更快。这个说法在单字段去重场景下其实站不住脚。你拿EXPLAIN分别看一眼SELECT DISTINCT uid FROM t和SELECT uid FROM t GROUP BY uid绝大多数 Spark 版本生成的物理计划几乎一模一样都是HashAggregate Exchange HashAggregate。Spark 的优化器在遇到 distinct 时会把它转换成和 group by 相同的聚合逻辑。所以如果你只是把DISTINCT字面替换成GROUP BY但业务逻辑和过滤条件完全没变性能上通常不会有什么实质提升。真正能带来提升的是改变去重粒度或者减少参与去重的数据量。比如需求是统计每天活跃用户数你原本在明细表上做COUNT(DISTINCT uid)改成先按(day, uid)粒度聚合一天只保留一条再做COUNT(DISTINCT uid)这一步已经把很多重复记录消掉了shuffle 数据量小了优化才真实生效。2.2 多字段 count distinct 才是重灾区比单字段 distinct 更容易踩坑的是同一张表里同时统计多个维度比如SELECT COUNT(DISTINCT uid) AS uv, COUNT(DISTINCT device_id) AS device_cnt, COUNT(DISTINCT order_id) AS order_cnt FROM detail_table WHERE dt 2024-06-01这条 SQL 表面上只有一个聚合节点但 Spark 为了分别对 uid、device_id、order_id 去重通常需要按这三个不同 key 分别做 shuffle。也就是说上面一条 SQL 可能触发三次全量数据重排。这也是很多单条 count distinct快、但多条 count distinct慢到离谱的根本原因。这种情况下一个常见优化是拆成多个 group by再用 join 合并结果-- 第一步分别算出各维度的去重计数 WITH uv AS ( SELECT COUNT(1) AS uv FROM ( SELECT uid FROM detail_table WHERE dt 2024-06-01 GROUP BY uid ) ), device AS ( SELECT COUNT(1) AS device_cnt FROM ( SELECT device_id FROM detail_table WHERE dt 2024-06-01 GROUP BY device_id ) ), order AS ( SELECT COUNT(1) AS order_cnt FROM ( SELECT order_id FROM detail_table WHERE dt 2024-06-01 GROUP BY order_id ) ) SELECT uv.uv, device.device_cnt, order.order_cnt FROM uv CROSS JOIN device CROSS JOIN order严格来说这个改写不一定每次都比原来快。因为拆分后明细表可能被扫描三次如果表非常大反而更慢。更合理的做法是尽量复用输入把明细表先缓存或落到临时表让三次 group by 都从同一个临时表读取减少重复 scan 成本。你还可以用spark.sql.optimizer.distinctKeys之类的优化项控制行为但不同版本表现有差异别盲信。另外提醒一个容易忽略的点如果要统计的是COUNT(DISTINCT col1, col2)也就是两列联合去重不要图省事用CONCAT(col1, col2)去拼。如果 col1 和 col2 里本身带分隔符或者值为 null拼接结果很容易撞车导致去重计数变小。正确做法是直接使用多列 distinct 语法或者用struct(col1, col2)作为去重键。3. 换一种写法业务上其实不需要 distinct3.1 取最新/最早一条记录用 row_number而不是 distinct有一种很典型的误用为了拿到每个用户最新的一条订单记录有人会先对用户和订单做 distinct然后 join 回原表。实际上用窗口函数一步就能做完WITH ranked AS ( SELECT uid, order_id, order_time, ROW_NUMBER() OVER (PARTITION BY uid ORDER BY order_time DESC) AS rn FROM detail_table WHERE dt 2024-06-01 ) SELECT uid, order_id, order_time FROM ranked WHERE rn 1如果你只需要每个 uid 一个最新order_id但订单量很大还可以先按(uid, order_id)聚合拿到最新的order_time再做一次ROW_NUMBER。第一次聚合能大幅减少进入窗口函数的行数代价更可控。窗口函数虽然也有 shuffle但它是按PARTITION BY uid分区的每个 uid 的数据会集中到同一个分区内做排序不会出现全表数据先按 order_id 去重散开再按 uid 重新聚集这种来回倒腾的情况。很多人没意识到distinct 加 join 的原方案其实是做了两次无关的 shuffle而窗口函数通常只要一次。3.2 判断存在性用半连接而不是 distinct 后再 join业务上有一种常见的子查询SELECT ... FROM a WHERE uid IN (SELECT DISTINCT uid FROM b)。你原本可能觉得先把 b 里的 uid 去重再去和 a 关联是正确写法。但对 Spark 来说去重后再 join 意味着 b 要完整执行一次 shuffle 去重a 和 b 又要做一次 join shuffle。其实IN子查询在 Spark 中会被优化成Left Semi Join它根本不需要预先去重。你直接写SELECT a.* FROM a LEFT SEMI JOIN b ON a.uid b.uid半连接只要能匹配到任意一条就直接输出 a 的行压根不需要先把 b 的 uid 全部去重再比对。这个改写通常能把先 distinct 再 join的两个 stage 压缩成一个 stage。同样的道理如果想找在 a 但不在 b 的用户用LEFT ANTI JOIN也会比b 去重后做 left join 再 filter清爽很多。判断条件是你到底需不需要 b 的明细如果只需要存在/不存在半连接和反连接永远优先于 distinct 加 join。3.3 先过滤、先聚合永远比先 distinct 更优无论怎么写有一条原则不会变参与去重的行数越少任务越快。所以优化 distinct 的第一步不是换算子而是检查能不能在去重前把数据压下去。常见手段有利用分区字段过滤只读必要分区在子查询里先做细粒度聚合比如先按(uid, device_id)去重再按uid去重如果明细表里同一 uid 有大量重复的辅助字段可以先把辅助字段 trim、清洗避免因为一两个空格导致去重失效把能下推的过滤条件下推到 scan 层让 Spark 读更少的数据。我自己实战中处理过一个任务原本COUNT(DISTINCT uid)前没有过滤数据源里有大量测试用户加了一个flag 1的过滤条件后输入直接从三亿行降到了几千万行耗时下降 80%。这种优化不 fancy但在很多场景下性价比最高。4. 数据倾斜下的两条实战优化路线4.1 单 key 极端倾斜加盐 两阶段去重假设uid分布极不均匀前 1% 的 uid 占了 70% 的行数。此时按 uid 做 hash 分区必然有一个 task 扛下大部分数据。直接调大分区数效果有限因为倾斜的 key 无论怎么分都会落在同一个分区里同一个 key 的 hash 是固定的除非加盐改变 key 本身。加盐的思路是给去重键拼接一个随机后缀让原本相同的 key 被打散到多个分区先在盐桶内完成第一层去重再去掉盐后缀做第二层精确去重。以 DataFrame API 为例import org.apache.spark.sql.functions._ val salted df .withColumn(salt, concat_ws(_, col(uid), (rand() * 10).cast(int))) // 第一层按 (salt, uid) 粒度去重注意这里 uid 和 salt 一起参与分组 val partial salted .groupBy(salt, uid) .agg(count(*).alias(cnt)) // 第二层去掉 salt按 uid 精确汇总 val result partial .withColumn(clean_uid, split(col(salt), _)(0)) .groupBy(clean_uid) .agg(sum(cnt).alias(real_cnt))这里有一个我踩过挺多次的坑第二层汇总一定要用sum(cnt)不能再用count(1)。第一层已经按(salt, uid)去重了同一 uid 在不同盐桶下的记录会各留一行第二层如果直接count就会把同一个 uid 在不同盐桶下的多条记录重复计算导致结果比真实值大 N 倍。加盐份数越多这个坑越隐蔽。如果你是在 Spark SQL 里写逻辑对应如下SELECT clean_uid, SUM(cnt) AS real_cnt FROM ( SELECT split(salted_key, _)[0] AS clean_uid, cnt FROM ( SELECT concat(uid, _, cast(rand() * 10 AS int)) AS salted_key, COUNT(*) AS cnt FROM detail_table GROUP BY salted_key ) t1 ) t2 GROUP BY clean_uid加盐份数通常取 10 到 100 之间不是越大越好。盐太多会导致 Map 端局部聚合失去意义中间结果膨胀盐太少又起不到均匀打散的效果。可以先看倾斜 key 的占比倾斜越严重盐桶数可以适当多些。4.2 多字段分别 count distinct拆到明细粒度之外如果倾斜发生在多个字段上比如既要 count distinct uid又要 count distinct device_id两个字段都倾斜上述加盐写法要分别做两套会比较麻烦。一个折中方案是先结合业务口径分析是否真的需要两个指标的精确值。如果只需要知道用户量级和设备量级近似去重往往是更好的选择详见下一节。如果业务要求精确建议把最耗时的那个指标先单独拎出来做加盐两阶段另外的指标可以走单独聚合然后 join。实践中我不建议在一张超大表上做太多花式改写因为每多一次 join 和 shuffle整个任务的稳定性都会下降。宁可多做几个中间临时表也不要在一个超级复杂的 SQL 里堆各种优化技巧出了问题极难排查。5. 近似去重业务能接受误差就用 HyperLogLog5.1 approx_count_distinct 的成本与误差Spark SQL 里有一个函数approx_count_distinct底层是基于 HyperLogLog 的近似基数估计。它不会把每个 key 都存下来而是维护一个固定大小的寄存器数组通过 hash 分布估算基数。优点是内存占用极小不需要为维护完整 key 集合付出巨大代价缺点是不精确。默认写法SELECT approx_count_distinct(uid) AS approx_uv FROM detail_table这里没传第二个参数时rsd默认是 0.05也就是相对标准误差大约在 1.6% 左右。你可以主动控制精度SELECT approx_count_distinct(uid, 0.01) AS approx_uv_precise FROM detail_tablersd越小越精确但内存和耗时也会上升。一般日常指标用 0.05 就够了如果用来做对账、风控阈值建议 0.01 或者干脆走精确计算。我在实际项目中用过一个很实用的组合先用approx_count_distinct快速算出一个数量级如果结果离业务阈值很近再对这部分数据走精确COUNT(DISTINCT)复核。这样既避免了大查询每次都跑几十分钟又保证了关键结论的可靠性。5.2 用 Bloom filter 做预过滤除了近似计数还可以在精确去重前用 Bloom filter 把明显不存在的 key 过滤掉。Spark 的DataFrameStatFunctions提供了bloomFilterAPI你可以对已知的大表构建一个 bloom filter再用它过滤另一张小表里的 key从而减少 join 或 distinct 的输入量。简单说Bloom filter 是一个宁可错杀、不可放过的过滤器如果它说 key 不存在那一定不存在如果它说存在可能实际上并不存在需要进一步精确判断。这个特性很适合在大表和小表关联去重场景下先把大表不可能匹配到的数据排除掉。使用前记得评估 false positive 率。这个参数并不是越低越好fpp 越低bloom filter 占用的内存越大构建成本越高。一般设 0.01 到 0.05 之间比较常见。遇到超大 key 集合时bloom filter 比维护完整哈希集合要省得多。6. 白嫖新版本特性AQE、分桶和动态裁剪6.1 AQE 开启后shuffle 分区会自动收敛吗Spark 3.0 引入 AQEAdaptive Query Execution之后distinct 类的 shuffle 也能受益不少。核心参数是这几个spark.sql.adaptive.enabledtrue spark.sql.adaptive.coalescePartitions.enabledtrue spark.sql.adaptive.coalescePartitions.initialPartitionNum200 spark.sql.adaptive.coalescePartitions.minPartitionNum20 spark.sql.adaptive.advisoryPartitionSizeInBytes64MBAQE 会在 shuffle 结束后根据实际 map 端输出的数据量动态合并小分区。比如你的 distinct 字段基数很低200 个分区里很多分区只有几千行AQE 会自动把分区数降到几十个减少后续 reducer 的调度开销整体速度会有明显提升但如果你开启 AQE 后仍然看到一个极端倾斜的 task它只能帮你合并小分区不能直接解决大 key 倾斜。倾斜场景还是得靠加盐这类算子级优化。另一个值得关注的是spark.sql.adaptive.skewJoin.enabled。它主要优化 join 倾斜如果你的 distinct 是通过 join 来做的这个参数会起作用但单纯对GROUP BY的倾斜没用别指望它能替代加盐方案。我的经验是新版本默认开启 AQE老版本一定要显式打开然后在 SQL 前加上一句注释记录参数方便后来人排查。6.2 分桶和预先重组让去重不再 shuffle如果某个去重场景每天都要跑而且 uid 的分布相对固定可以提前把表按uid分桶。分桶的好处是同一 uid 的所有数据在物理存储上已经落到同一个桶文件后续GROUP BY uid时Spark 可以直接在桶内聚合避免再次全量 shuffle。CREATE TABLE dim_user_bucketed ( uid STRING, ... ) USING parquet CLUSTERED BY (uid) INTO 64 BUCKETS不过分桶不是银弹。桶数要和你后续查询的 shuffle 分区数匹配如果查询时并行度远超桶数或完全不匹配优化效果会大打折扣。另一个问题是数据更新频繁时维护分桶成本很高。我比较推荐把它用在稳定的中间层表或者维度表而不是每天凌晨疯狂刷新的明细表。如果你只是偶然跑一次去重临时去分桶就是画蛇添足。6.3 动态分区裁剪减少输入量如果你的去重查询总是要关联另一个分区表比如detail_table按dt分区、用户维表按日更新那么spark.sql.optimizer.dynamicPartitionPruning.enabledtrue可以让 Spark 在运行时根据小表结果动态裁掉大表中不需要的分区。它不影响 shuffle但可以直接减少 Scan 阶段读入的数据量间接让 distinct 更快。这个特性最管用的场景是大表和小表 join 后再去重。以前你可能需要先JOIN两张大分区表再DISTINCT。开启动态分区裁剪后Spark 会先读小表把小表涉及的分区列表算出来再裁剪大表的读入范围省下的是 IO 和时间成本。很多人的 distinct 慢其实不是 distinct 本身慢而是它前面那个大 join 把数据放大了。剪掉无关分区问题自然缓解。7. 一次模拟项目的去重优化复盘7.1 原始任务长什么样为了把这套方法串起来我拿一个模拟项目举例。某业务表user_action_log每天新增约 3 亿条行为日志字段包括uid、device_id、action_type、log_time等。需求是统计当天活跃用户数和活跃设备数SQL 原本长这样SELECT COUNT(DISTINCT uid) AS active_user_cnt, COUNT(DISTINCT device_id) AS active_device_cnt FROM user_action_log WHERE dt 2024-06-01第一次跑任务卡了 12 分钟其中一个 stage 的 task 有严重的磁盘溢写。看EXPLAIN能看到两个Exchange hashpartitioning一个按 uid一个按 device_id而且 uid 本身倾斜严重top 100 个 uid 贡献了将近一半的日志量。7.2 三轮优化后的效果第一轮我先把过滤条件做扎实。业务方其实只关心action_type IN (login,purchase)的行为日志里还有大量浏览行为。加了过滤后参与计算的行数从 3 亿降到了 1.6 亿耗时从 12 分钟降到 6 分钟。这一步没有用任何高级技巧就是减少了输入数据量。第二轮业务方说活跃用户数可以接受 1% 到 2% 的误差于是把COUNT(DISTINCT uid)改成approx_count_distinct(uid, 0.02)device 同理。耗时从 6 分钟降到 45 秒。同时我保留了每周一次全量精确对账的任务用精确结果和近似结果做差确保误差在可控范围内。第三轮为了满足另一条必须精确的口径我针对 uid 做了加盐两阶段优化。盐桶数取 20第二层用sum(cnt)而不是count(1)精确版任务跑到了 2 分钟左右。对比原始 12 分钟已经算很大幅度的提升了。如果再把 AQE 打开合并小分区整体还能再往下压一点。7.3 优化过程中踩过的坑这次改造里踩了三个值得记录的坑。第一个就是加盐后第二层聚合用错函数。我前面已经强调过第一层按(salt, uid)去重后第二层必须sum(cnt)。实际开发中因为直觉上去重就是 count distinct很容易顺手写个count(*)结果算出来比真实值多出十几万。排查半天才发现是盐桶导致的重复计数。第二个坑是过滤条件下推并没有自动生效。起初我把action_type IN (login,purchase)写在最外层查询的 where 里但明细表太大Spark 并没有把过滤条件很好地推入到前面子查询的 scan 阶段导致还是全表扫。后来手动把条件提到内层子查询里输入数据量才真正降下来。所以不要迷信优化器关键过滤条件下推不了时人工调整 SQL 顺序非常有必要。第三个坑是approx_count_distinct的误差不是稳定的少算。HyperLogLog 的误差是随机波动可能多也可能少不能只看一次结果就放心。我在项目里固定每天跑一次精确抽样对比连续观测一周确认误差在业务可接受范围内才把近似结果作为默认指标。最后再分享一点个人体会接手任何 distinct 优化需求我第一步永远不是动手改 SQL而是先确认业务到底要什么。是要精确值还是近似值是只要数量还是要明细能不能接受把去重粒度放宽一天这些问题问清楚了很多优化手段的取舍就顺理成章了。Spark 的执行引擎再智能也架不住业务口径本身模糊带来的反复重跑提前对齐口径比省下的那几分钟更值钱。