Spark SQL的distinct我愿称之为最容易被低估的算子。表面上看就是一个关键字SELECT DISTINCT user_id FROM logs翻译成人话就是“把重复的去掉”听起来毫无技术含量。但凡用Spark跑过TB级以上数据的人都会知道distinct可能是整个任务里最贵的一步一次全局shuffle、一次map端局部聚合加一次reduce端聚合所有候选值都得跨网络走一遍你还不能随便改。如果SQL里再出现两三个COUNT(DISTINCT ...)那基本可以提前规划加班了。这篇文章我从执行原理讲到实操优化最后附一个把UV统计从50分钟压到9分钟的完整案例适合天天和Spark SQL打交道的数据开发、数据仓库工程师也适合刚接触Spark想搞明白“为什么我一条去重SQL能跑这么久”的同学。先说结论distinct在Spark里从来不是“过滤”它是一个如假包换的全局限聚合操作。1. 先说结论distinct在Spark SQL里到底是怎么执行的很多人一提到distinct脑子里浮现的是Excel删除重复项一行行比对、删掉多余的。Spark里完全不是这套逻辑分布式系统没法把所有数据堆到一台机器上慢慢比它必须让所有“相同的值”在shuffle之后落到同一个分区才能知道哪些重复哪些不重复。所以distinct本质上是一个按去重列进行分组、且不带任何聚合函数的Aggregate操作。Catalyst优化器会把SELECT DISTINCT a, b改写成Aggregate [a, b]物理计划就是分阶段聚合加一次shuffle。理解这个前提后面所有优化思路才立得住。1.1 一条最简单的distinct SQL物理计划长什么样先看一个再普通不过的查询EXPLAIN SELECT COUNT(DISTINCT user_id) FROM behavior_log WHERE dt 2024-06-01;Spark返回的物理计划会是这样版本和配置不同会有细节差异但骨架一致 Physical Plan HashAggregate(keys[], functions[count(1)]) - HashAggregate(keys[user_id#12], functions[]) - Exchange hashpartitioning(user_id#12, 200) - HashAggregate(keys[user_id#12], functions[]) - FileScan parquet ... [dt2024-06-01]我们从下往上看最内层HashAggregate在map端做局部去重。每个executor只看到自己处理的那部分数据先把相同user_id合并掉。Exchange hashpartitioning按user_id的哈希值做shuffle把所有相同的user_id送到同一个reduce端分区。这一行是整个distinct最贵的部分数据量越大shuffle越伤。中间层HashAggregatereduce端收到各map端送来的局部去重结果再做一次最终去重。到这里输出的已经是全局唯一的user_id集合。最外层HashAggregate对唯一集合执行COUNT(1)。所以你看count distinct比普通聚合多的不是一点点它相当于先做了一次完整的分组去重再做一次计数两个聚合算子中间还必须夹一次全网状shuffle。1.2 为什么count(distinct)比普通聚合慢一个量级拿SELECT COUNT(*) FROM logs来对比这个查询其实根本不需要shuffle。每个分区在map端各自数一遍行数输出一个“本分区行数”这样的单行中间结果reduce端把几百个数字加起来完事。整个shuffle传输的数据量是“分区数 × 8字节”在TB级数据面前几乎可以忽略。COUNT(DISTINCT user_id)就不一样了。map端局部聚合输出的是“本分区出现的所有不同user_id”这个集合有多大如果数据重复率低它可能和输入行数差不多。也就是说shuffle阶段要传输的几乎是一整列user_id几十亿行字符串在网络里满天飞。同样的任务规模和集群配置distinct的shuffle数据量可能比普通聚合高出几个数量级。还有一个经常被忽略的问题去重列如果基数非常高reduce端的HashAggregate需要维护一个巨大的哈希表来保存所有唯一值。这个状态内存一旦撑不住就会溢写到磁盘或者退化成更慢的排序聚合。我见过不少任务不是死在shuffle上而是死在reduce端聚合状态把executor内存打爆上。1.3 distinct、group by、窗口去重的物理计划差异我把日常里最常见的三种“去重”写法拉出来对比一下很多人以为它们是同一回事物理计划差别其实很大。写法Catalyst实际改写物理计划特征适用场景SELECT DISTINCT user_id FROM tAggregate(user_id)一次Exchange 两次HashAggregate只要去重结果集SELECT user_id FROM t GROUP BY user_idAggregate(user_id)同上物理计划几乎一致同上习惯用group by的人更多ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY dt)Window需要全量排序或rangepartition通常最贵要保留每个user_id某一条明细比如取最新一条DISTINCT和GROUP BY在Spark SQL里基本是同一套物理执行流程所以不存在“用group by替代distinct就更快”这种说法优化器早就帮你改写好了。但窗口函数去重是真贵ROW_NUMBER()虽然能拿到明细代价是要对所有窗口数据进行排序或带状态的流式处理在能接受“只去重、不要明细”的场景里尽量别用它。2. 优化的第一层把“精确去重”换成“近似去重”聊优化我先说一个不太政治正确的经验很多distinct任务根本不需要精确结果。产品看板上的DAU少个几百、大屏上的活跃用户数差个千分之一肉眼根本看不出来。但你要是拿这个理由去跟业务方说“我想把精确去重改成近似算法”多半会被打回来因为没人愿意为千分之几的误差背锅。所以在动SQL之前先搞清楚需求的真实口径是财务结算、对账这类一分钱不能差的还是趋势分析、规模估算这种误差允许在几个百分点内浮动的。后者直接用approx_count_distinct性价比极高。2.1 approx_count_distinct与其背后的HyperLogLogapprox_count_distinct基于HyperLogLogHLL算法。它的核心思想很巧妙对每个值做一次哈希得到一个二进制串然后看这个二进制串从末尾起连续有多少个0或者从开头起连续多少个1。如果集合里有N个不同的值理论上你就有N次独立随机的“抛硬币”机会最长的那串0的位数可以直接估算出N的数量级。单个值只需要保存它哈希后的那点信息不需要保存值本身。在Spark SQL里用法极其简单SELECT approx_count_distinct(user_id, 0.01) AS uv FROM behavior_log WHERE dt 2024-06-01;第二个参数0.01是期望的相对标准差官方默认是0.05也就是5%的误差范围。你给的精度越高HLL需要的寄存器数组越大内存占用越高但即使是很高的精度也就是KB级别的内存和精确去重维护一个完整哈希表完全不在一个量级。2.2 误差换算与真实业务场景取舍假设真实UV是1亿用默认的5%误差结果会在9500万到1.05亿之间。这个波动对日常看板可以接受但如果业务方拿这个数字去做投放结算误差几百万人那肯定不行。我实际踩过一个坑某个项目把DAU指标从精确改成了近似上线一周没人发现后来有一天活动运营根据这个数字做预算差了几万人业务侧炸了。所以结论很清晰——适合近似用户规模趋势、渠道对比、离线报表的中间过程、预研性分析。必须精确结算、对账、需要和财务数据交叉验证的指标、任何写进合同的数字。还有一点容易被忽略approx_count_distinct的结果是不稳定的。同一份数据每次跑因为分区、executor调度的微小差异HLL合并顺序不同最终数字会有波动。如果业务方是那种“为什么这次比上次少了5000”都要来质问你的类型趁早别用。2.3 近似去重的三种正确用法我建议这样用而不是一个“换”字了事第一日常看板和大屏类指标直接用近似把“误差范围”这个预期提前跟业务对齐别等上线后让人发现数字会跳。第二跑精确任务之前先用近似结果做冒烟验证。比如预期UV是百万级近似跑出来突然变成千万级那多半是数据质量出问题了比如某些字段被人为注入了大量唯一ID。这时候直接省去一次精确任务的半小时等待。第三在资源预算、容量规划场景下用近似值估算基数决定后续分桶数、join策略和存储格式选型。精确值在这种决策场合没有意义近似值完全够用。3. 优化的第二层改写SQL减少shuffle次数如果说近似去重是在“结果精度”上做文章那这一层就在“执行代价”上做文章。核心目标只有一个让shuffle的数据量尽量小让shuffle的次数尽量少。distinct相关SQL写得不讲究最容易出现的问题是多个COUNT(DISTINCT)堆在一个SQL里以及distinct和join的顺序放错。这两条都是新手重灾区。3.1 多个count distinct必须拆开的三个理由很多人习惯这么写SELECT COUNT(DISTINCT user_id) AS uv, COUNT(DISTINCT device_id) AS dv, COUNT(DISTINCT order_id) AS ov FROM orders WHERE dt 2024-06-01;看起来是一个查询三个指标很简洁但执行起来是个灾难。Spark的优化器遇到多个distinct聚合时会为每个distinct列分别生成聚合路径再通过join把结果拼起来本质上是把原SQL拆成了相当于三份独立去重的计算量再额外加了一次结果归并的代价。而且这三路计算在同一个stage里串成一串任何一个distinct列倾斜整个任务都得等它。第一每个distinct列都要完整shuffle一次自己的原始数据三列就是三倍网络开销。第二多个distinct的中间结果如果通过sort merge join连接又会引入额外的shuffle和排序。第三几个distinct列的基数在组合极端情况下还会造成中间膨胀。改进方案也直白——拆开并行跑from concurrent.futures import ThreadPoolExecutor def query_count(sql): return spark.sql(sql).collect()[0][0] jobs { uv: SELECT COUNT(DISTINCT user_id) AS cnt FROM orders WHERE dt2024-06-01, dv: SELECT COUNT(DISTINCT device_id) AS cnt FROM orders WHERE dt2024-06-01, ov: SELECT COUNT(DISTINCT order_id) AS cnt FROM orders WHERE dt2024-06-01, } with ThreadPoolExecutor(max_workers3) as pool: results {name: pool.submit(query_count, sql) for name, sql in jobs.items()} final {name: future.result() for name, future in results.items()}三个独立任务并发跑互不拖累比塞在同一个SQL里串行执行稳定得多。不过这招的前提是资源足够集群资源本来就很紧张的时候三路并发会把资源瓜分得更碎反而更慢需要自己权衡。3.2 先过滤再distinct谓词下推与预裁剪这句话听起来像废话但实际执行中特别容易被忽略。就拿开头的SQL举例SELECT COUNT(DISTINCT user_id) FROM behavior_log WHERE dt 2024-06-01;先确认一下这是不是真的只读了需要的分区和列。很多表是按dt分区没错但如果你忘了写分区条件或者写成了WHERE dt 2024-06-01 AND dt 2024-06-01却因为字段类型是字符串导致边界判断出问题Spark会把全表分区都扫进来distinct基数瞬间膨胀几十倍。除了分区裁剪还要注意列裁剪。COUNT(DISTINCT user_id)理论上只需要user_id这一列但实际线上表往往有几十个字段。虽然Parquet有列式存储的优势扫描时会自动做列裁剪但如果你前面的子查询为了省事写了SELECT *再在外层做distinct那扫描和shuffle的可能就是整个宽行代价完全不同。正确的姿势是在子查询里只select你真正需要的去重键把一切能提前过滤的条件事件类型、渠道、时间范围都放到最内层让Spark的谓词下推在最源头起效。比如SELECT COUNT(DISTINCT user_id) AS uv FROM ( SELECT user_id FROM behavior_log WHERE dt 2024-06-01 AND event_type login AND channel IS NOT NULL ) t这个改写没有引入任何新技术纯粹是让你“少给Spark喂数据”但往往立竿见影。3.3 distinct join的顺序怎么摆distinct和join放一起顺序是决定性能的关键。很多人习惯把join结果集直接丢给distinct看着没问题实际白白多shuffle了无数重复行。-- 差写法先join出宽表再distinct SELECT DISTINCT u.user_id, u.channel FROM orders o JOIN dim_user u ON o.user_id u.user_id WHERE o.dt 2024-06-01orders表里一个user_id可能对应几十条订单join之后user_id会重复几十甚至上百次然后distinct再去把这些重复消掉。你花大量资源把重复行join出来又花大量资源把它们shuffle着消掉纯属白做功。更好的写法是先在orders端把user_id去重把“需要join的候选集合”压缩到最小再去做joinSELECT DISTINCT u.user_id, u.channel FROM ( SELECT DISTINCT user_id FROM orders WHERE dt 2024-06-01 ) o JOIN dim_user u ON o.user_id u.user_id这个顺序变化背后的核心原则是distinct要shufflejoin也要shuffle那就要尽量让两条路径上的数据都更小。先把orders的大表压成“当天不同user_id的小集合”join阶段的shuffle和distinct阶段的shuffle就都变小了。同样的道理如果只统计某个渠道的用户还可以在join前就把维度表也过滤一遍。4. 优化的第三层处理数据倾斜与资源不均很多distinct任务在SQL层面看着没毛病分区过滤做了、重复率低、网络传输量也不大但就是慢得像老牛爬坡。这种情况十有八九是数据倾斜。一个现象就能判断Spark UI上某个stage里九十九个task几秒就跑完了剩下一个task要跑二三十分钟整个任务就卡在这个task上。4.1 大key识别与双管道处理distinct场景下的倾斜本质是有个别超高基数的key比如某个头部用户的日志量是普通用户的几千倍。这些用户的数据在shuffle时被hash到同一个分区那个分区的reducer要处理海量数据而其他分区早就空闲了。处理思路其实并不复杂把高频key和普通key拆开来跑。第一步先按user_id聚个类看哪些key的记录数远超正常水平SELECT user_id, COUNT(1) AS cnt FROM behavior_log WHERE dt 2024-06-01 GROUP BY user_id HAVING cnt 100000把命中的高频user_id集合缓存下来比如广播变量。然后对原表做标记高频的一组走专门优化的通道低频的一组走常规distinct最后把两边的UV数直接相加。因为两边是互斥集合各自去重后加起来就是全量去重数这个结果是精确的。但更通用、代码也更省事的方案是加盐二阶段聚合我直接给一个完整示例WITH salted AS ( SELECT user_id, CAST(RAND() * 16 AS INT) AS salt FROM behavior_log WHERE dt 2024-06-01 ), level1 AS ( SELECT user_id FROM salted GROUP BY user_id, salt ) SELECT COUNT(1) AS uv FROM ( SELECT user_id FROM level1 GROUP BY user_id ) t这个方案的原理是第一次shuffle按(user_id, salt)这个复合键分区同一个高频user_id因为带了16种随机盐值会被打散到16个分区任何一个reducer都不会被单个key压垮。第一次聚合输出的是全局唯一(user_id, salt)对此时一个user_id最多对应16条数据量已经从“重灾区”降成“小水池”。第二次shuffle再按user_id去重每个user_id最多16行分区负载完全可控。值得注意的是这个方案有个前提原始数据必须存在大量重复加盐才有实际收益。如果一个user_id本来就只有一行记录加盐反而会把一行变16行平白多了一次shuffle。所以在对重复率没底的时候先看一眼map端聚合输出和输入的行数比再决定要不要用这招。4.2 AQE能做和不能做的事从事Spark 3.2开始Adaptive Query ExecutionAQE默认是开启的它的核心能力是在shuffle完成之后、根据实际产生的数据分布来动态调整执行计划。对distinct任务来说AQE能帮两件事第一件事是自动合并小分区。如果你设置了很高的spark.sql.shuffle.partitions但数据量其实没那么大shuffle结束会产生大量小分区每个分区数据只有几MB纯纯的调度开销。AQE会把小分区合并成合理大小减少task数量。第二件事是运行时调整join策略。distinct前面如果接了joinAQE可能把原本的sort merge join动态降级成broadcast join把右表广播到所有executor省掉一次大shuffle。但AQE有个明显的边界它不能直接解决聚合阶段的单key倾斜。skewJoin处理的倾斜是join过程中某个分区数据量过大而对独立的HashAggregateAQE没有那种“把倾斜聚合状态拆开”的能力。所以前面提到的加盐方案该手写还得手写AQE替代不了。常用的AQE相关配置我贴一份仅供参考SET spark.sql.adaptive.enabledtrue; SET spark.sql.adaptive.coalescePartitions.enabledtrue; SET spark.sql.adaptive.advisoryPartitionSizeInBytes256MB; SET spark.sql.adaptive.skewJoin.enabledtrue; SET spark.sql.adaptive.skewJoin.skewedPartitionFactor5; SET spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes256MB;spark.sql.shuffle.partitions默认是200对大表来说常常不够对中小表又太多。一个我常用的经验是让shuffle分区数约等于集群可用core数的2到4倍同时配合AQE的advisoryPartitionSizeInBytes让数据量自己决定该有多少个reducer而不是拍脑袋定个数。4.3 分桶表、布隆过滤等进阶手段分桶表bucketed table对distinct的收益是间接的。如果表本身已经按照去重键做了分桶数据在物理存储层面就相对均匀地打散在集群里扫描阶段的负载分布会比普通表更健康。但要注意分桶并不等于预聚合Spark对distinct的全局唯一性要求仍然需要通过shuffle来保证所以别指望它能直接免掉Exchange。布隆过滤器则适合另一种场景用一个很小的集合去裁剪一个超大集合。布隆过滤器有一个核心特性它永远不会漏报只有可能误报。也就是说如果布隆说“这个值一定不在集合里”那它就真的不在如果它说“可能在”那只是可能需要进一步确认。基于这个特性你可以先对某个候选集合比如一批已知投放用户构建布隆过滤器然后在海量日志表里把“一定不在这批用户里”的行安全地过滤掉剩下的行数量会小一个量级再去做精确的distinct计数精准且无误差。比如import org.apache.spark.util.sketch.BloomFilter // 从候选集构建布隆estimatedNumItems表示期望元素数fpp表示误判率 val bloom BloomFilter.create[String](10000000L, 0.01) candidateDF.collect.foreach(row bloom.putString(row.getString(0))) val broadcastBloom spark.sparkContext.broadcast(bloom) // 用布隆过滤掉必然不存在的行 val filteredDF logDF.filter( row broadcastBloom.value.mightContainString(row.getString(0)) )注意这里布隆过滤器只是个预过滤角色最终计数仍然用精确的count distinct。布隆的存在只是为了把“绝对无关的数据”在shuffle之前删掉相当于一个聪明版的谓词下推。5. 实操案例一次UV统计任务从50分钟压到9分钟理论聊了这么多我拿一个真实处理过的任务把整个优化过程串一遍。这个案例我稍微改了改业务字段但数据规模和优化链路是照实的。5.1 原始SQL与执行瓶颈业务背景某产品团队要统计每天的去重用户数UV所谓UV就是当天访问过产品的不同用户数量。数据来源是一张行为日志表日增量约1.2亿行上线跑批任务时任务运行50分钟业务侧天天抱怨。最初的SQL特别简单SELECT COUNT(DISTINCT user_id) AS uv FROM behavior_log WHERE dt 2024-06-01;我先开了EXPLAIN看物理计划然后对照Spark UI定位到三个核心问题第一表里实际有几十个字段虽然SQL只用了user_id但因为表设计历史原因读取时文件列裁剪没有完全生效扫描阶段加载了很多无谓的数据。第二shuffle分区数用的默认200单分区数据量过大部分reduce端task的聚合状态压力很高。第三shuffle阶段有个别user_id的数据量异常大这就是前面说的倾斜问题有个别task拉取的数据量是中位数的几十倍。5.2 逐步优化与每一层的收益整个优化过程不是一步到位而是一层一层压下来的步骤操作内容任务耗时基线原始SQL50分钟第1步子查询只保留user_id字段加事件类型过滤42分钟第2步尝试approx_count_distinct快速估算6分钟但误差不达标第3步全量加盐二阶段聚合解决倾斜14分钟第4步调大shuffle分区数到800配合AQE自动合并9分钟第1步属于白捡的收益只做了列裁剪和提前过滤没动执行计划骨架就省了8分钟。第2步证明了近似结果确实很快从42分钟直接压到6分钟但业务方复核后发现预估误差超过了千分之一接受不了只能放弃。第3步和第4步是组合拳加盐方案把shuffle阶段最慢的那个task从几十分钟降到了正常水平而调大分区数配合AQE又让reducer的数量和数据量匹配上了最终稳定在9分钟。5.3 最终方案与可复用清单最终的精确方案长这样综合了列裁剪、加盐和参数调整WITH salted AS ( SELECT user_id, CAST(RAND() * 8 AS INT) AS salt FROM ( SELECT user_id FROM behavior_log WHERE dt 2024-06-01 AND event_type login ) ), level1 AS ( SELECT user_id FROM salted GROUP BY user_id, salt ) SELECT COUNT(1) AS uv FROM ( SELECT user_id FROM level1 GROUP BY user_id ) t这个SQL里盐值我选了8而不是16因为实测这个表的重复率没那么极端8路打散已经足够抹平倾斜再大反而增加二次聚合的负担。复盘整个优化过程我总结出一份可以反复参考的检查清单写任何distinct之前先EXPLAIN确认到底有没有走你预期的物理计划。扫描阶段一定要检查分区裁剪和列裁剪源头少读数据是性价比最高的优化。精确需求和近似需求分开不要拿精确任务的价格去跑近似场景。多个COUNT(DISTINCT)坚决拆开独立指标独立跑。倾斜严重就在去重键上加盐做两阶段聚合。分布式参数不是拍脑袋定的观察Spark UI里每个task处理的数据量再调。6. 常见问题与排查速查表做数据开发总会遇到一些“看起来没毛病但就是跑不动”的distinct任务。我把日常排障遇到的高频问题整理成一张速查表遇到可以直接对着查。表现可能原因排查思路解决方案任务卡在shuffle阶段某个task远超其他task单key数据倾斜Spark UI看Exchange阶段各task的shuffle read量加盐二阶段聚合或高频key分流executor频繁OOM或GC严重去重列基数高聚合状态过大看executor峰值内存和GC日志调大分区数降低单分区数据规模检查是否可转为近似算法distinct结果偏小或每次跑都不一样用了approx_count_distinct核对业务口径是否为精确需求精确场景换回count(distinct)近似场景向业务说明波动范围SELECT COUNT(DISTINCT a, b)报语法错误版本不支持多列count distinct写法查看Spark版本与语法支持用COUNT(DISTINCT STRUCT(a, b))或子查询group by a, b再count一个查询里多个count distinct整体极慢每个distinct都要独立shuffle再join看物理计划是否出现多个AggregateJoin拆成独立任务并行跑或改用近似算法数据量不大但distinct很慢扫描了多余分区/列EXPLAIN看FileScan的PartitionFilters补齐分区过滤子查询只select去重键distinct后数值比预期大join导致行数膨胀被去重的集合变大检查distinct和join顺序先对事实表distinct再join或先过滤维度表6.1 执行计划看不懂的几行英文很多同学第一次看distinct的执行计划会被HashAggregate、ObjectHashAggregate、SortAggregate这几个词搞晕。简单区分HashAggregate是目前主流的哈希聚合实现快但需要内存容纳聚合状态ObjectHashAggregate处理复杂类型的聚合键更通用但更吃内存SortAggregate是退路当哈希表溢出内存时会基于排序实现慢但稳定。看到执行计划里的Exchange hashpartitioning那一行就是shuffle的位置它是整个任务里最值得盯紧的瓶颈点。PartitionFilters那一行则显示了扫描阶段实际生效的分区过滤条件如果它是空或者只显示[]说明你的分区裁剪没生效数据可能全表扫描了。6.2 distinct结果疑似不对怎么排查先确认有没有用了近似函数这是最常见的“结果不对”来源。其次要注意null的处理语义COUNT(DISTINCT col)是不计入null值的但GROUP BY col会把null当作一个正常组两边对null的语义不同如果你拿group by的count对不上distinct的count先从null下手排查。字符串字段的去重还容易踩隐式类型的坑比如同一个user_id有的行的值是10001有的是10001Spark在比较时会做隐式转换结果可能合并也可能是两组取决于字段声明类型。去重前统一字段格式是最省事的办法。如果distinct的结果异常偏大几乎都是因为前面join产生了冗余乘出来的重复行把集合撑大了。这时候不要盯着distinct本身调参回头砍join。6.3 日常写distinct时必守的五个习惯第一写每条distinct前至少跑一次EXPLAIN确认物理计划里有没有意外的大shuffle。第二能提前减量绝不等到distinct时再减过滤条件全部下沉。第三手上同时有精确和近似两套指标时先问业务到底看哪个别默认上精确。第四多个distinct指标别塞同一个SQL拆开跑更稳。第五运行中定期打开Spark UI看每个stage的数据分布倾斜往往在第一次跑就露出马脚越早发现越省事。我在处理类似问题的过程中还有一个感受就是distinct慢往往不是去重这一步本身慢而是它前面的join把数据撑大了、或者它后面对结果还有一轮大聚合。把整条链路拆开来看有时候真正要优化的不是distinct而是它的上下游。把问题和算子解耦再对症下药比一股脑调shuffle参数管用得多。希望这篇对你有帮助少走几次弯路。
