1 场景回顾与问题定位先聊聊我在实际工作中最常遇到的情况一张事实表几亿条甚至几十亿条记录每个用户每天可能产生多条行为记录业务方要做报表需要统计“独立用户数”或者从明细里找唯一键集合。大家最直接的反应就是写SELECT COUNT(DISTINCT user_id) FROM ...或者SELECT DISTINCT ...然后任务就挂在那边跑了半个小时还在 shuffle运气好赶上集群繁忙直接跑挂。可能有人会觉得DISTINCT 而已Spark 难道搞不定其实它能搞定但搞定的前提是你得知道它在幕后做了什么。Spark SQL 对 DISTINCT 的默认处理逻辑本质上是走HashAggregate或SortAggregate当数据量上来、分组维度多、单个 key 的基数又很大的时候shuffle 的数据量会非常夸张。我之前遇到过一张表有 300 多个字段只需要取其中 5 个字段做去重结果 Spark 把整行数据都拖到了 shuffle 阶段网络 IO 直接爆了CPU 也一直在做序列化和反序列化。这个项目标题是“Spark SQL 之 distinct 优化”我最初接到这个需求时业务那边反馈的痛点非常典型跑一次COUNT(DISTINCT)要 40 分钟以上而业务报表要求半小时内出数。集群资源经常被这个任务占满其他任务跑不动甚至触发资源队列死锁。数据量还在每天增长不加优化的话越往后越慢。如果你也遇到过类似的场景这篇文章的思路应该能帮你省下不少调优时间。下面我会从 distinct 的底层行为、优化思路、实际操作、问题排查四个层面展开全程基于我自己的实践经验写案例会用模拟数据说明不会牵扯任何真实业务细节。2 DISTINCT 底层执行逻辑深度拆解2.1 Spark 对 DISTINCT 的默认处理流程先说清楚一件事DISTINCT在 Spark SQL 里不是一个独立的物理算子而是被转换为Aggregate。以SELECT DISTINCT key1, key2 FROM table为例Spark 会做一次全局聚合分组字段就是 key1 和 key2没有任何聚合函数。这个过程分两个阶段局部聚合Partial每个 executor 上的 task 先对自己处理的分区做一次去重输出局部结果。全局聚合Final将所有 task 的局部结果按 key 进行 shuffle然后合并成一个全局结果集。如果写成COUNT(DISTINCT column)逻辑类似但不同的是它不仅要按 column 去重还要在 shuffle 过程里把 column 的值作为 key 传输。这里就出现一个很多人忽略的性能陷阱当 key 的基数特别高比如一亿个不同的 user_idshuffle 过程要传输一亿条记录每条还带着完整的 key 值网络和磁盘开销非常惊人。我亲眼见过一个场景DISTINCT 之后还要 JOIN 另一张维表结果 DISTINCT 输出的文件数量巨大拖到 JOIN 阶段直接引发了文件打开数超限的报错。这是因为 Spark 默认会在 DISTINCT 之后写 shuffle 文件如果 reduce task 数量设置不当就会生成海量小文件。2.2 常见的性能瓶颈藏在哪根据我的排查经验DISTINCT 慢的背后通常集中在四个位置数据倾斜某个 key 的记录数远超其他 key导致单个 reducer 长时间无法完成整个 stage 就卡在那里。shuffle 数据量过大默认情况下 shuffle 过程中参与分布式的列越多传输的数据量越大。如果表字段很多而你只对其中几个字段去重Spark 依然可能会将整行数据参与序列化这取决于spark.sql.parquet.mergeSchema等配置和具体的优化规则。序列化开销默认的JavaSerializerKryo 需要手动开启在对象较大时性能较差CPU 长时间消耗在序列化和反序列化上。spark.sql.shuffle.partitions 设置不合理200 是一个通用的默认值但在几亿数据量面前200 个 reduce 可能太少也可能太多需要根据数据量动态调整。2.3 为什么 ORDER BY 会让问题更复杂很多人在用 DISTINCT 时还会顺手加一个ORDER BY比如SELECT DISTINCT col FROM table ORDER BY col。一旦加了这个Spark 就会在全局聚合之后再加一个Sort阶段。这个 Sort 会重新触发一次全量 shuffle排序的数据量和 DISTINCT 输出的结果量一样大内存压力成倍增加。如果你真的需要去重后的有序结果不要在 SQL 层面同时写 DISTINCT 和 ORDER BY建议先去重写出临时表再对临时表做排序查询。这样能避免 Spark 在同一个查询计划里叠加两个高开销操作。我测试过把一个“DISTINCT 全局排序”的查询拆成两步整体耗时下降了大概 35%效果非常明显。3 优化前的关键判断3.1 从 SQL 执行计划判断瓶颈点拿到一条慢查询先别急着改 SQL第一件事是看执行计划。用EXPLAIN可以看到 Spark 的计划重点看两个东西Exchange节点的数量和数据分区方式。一条绝对的慢 DISTINCT 查询计划里通常会有一个巨大的HashAggregate后面跟着Exchange, 并且 Exchange 的hashExpressions是被去重列的完整表达式。如果一个查询里有多个 DISTINCT比如SELECT COUNT(DISTINCT a), COUNT(DISTINCT b) FROM tableSpark 的优化器在没有开启多 distinct 优化的老版本里会把每个 distinct 单独处理产生多次 shuffle。我在项目里就遇到过这种逻辑业务要同时计算多个维度的独立用户数一条 SQL 里写了四五个 COUNT(DISTINCT)跑了两个小时。我尝试开启spark.sql.optimizer.aggressiveAggregation相关的优化结果新版本和老版本行为还不一样这个下面会细讲。3.2 根据数据规模判断方案走向不同的数据量优化策略完全不一样。我的经验是这样百万级以内默认 DISTINCT 完全够用不需要额外优化。千万级到亿级建议先尝试调spark.sql.shuffle.partitions增加并行度再考虑改写 SQL。十亿级以上必须从业务角度考虑是否真的需要精确去重或者使用近似去重算法如 HyperLogLog或者预处理去重结果避免查询时大批量跑原始明细。我在实际项目中碰到的都是最后一种十亿级明细表每天早上业务要前一天的独立设备数量。原始 SQL 跑了一个半小时完全不能接受。3.3 业务需求是否需要精确去重这个点我觉得特别值得强调。很多场景下业务方要的“独立数”并不需要绝对精确比如报表里偏差 1% 完全不影响决策但团队内部为了“严谨”一直坚持用精确统计。其实这是成本浪费因为可以用近似算法把几十亿的量级压到几百万以内的计算量。Spark SQL 原生提供了approx_count_distinct函数基于 HyperLogLog 算法能达到 1% 左右的误差率。我把话放在这里如果你的业务需求允许误差在 1%~2%一律用approx_count_distinct你会回来感谢我的。4 实操优化方案拆解4.1 方案一预处理聚合减少扫描数据量这是最基础的一步但很多人会忽略。假设你每天需要计算“活跃用户数”而原始订单表是流水粒度的同一个用户可能一天内有几十条流水。如果直接对原始流水表 COUNT(DISTINCT user_id)你需要扫描整个流水表算力全部浪费在重复数据上。正确做法先按用户粒度做聚合把同一个用户当天所有记录压缩成一条再对聚合结果去重。用代码表示CREATE TABLE tmp_user_daily AS SELECT user_id, date, MAX(order_id) AS last_order_id FROM raw_order_table WHERE date 2024-01-01 GROUP BY user_id, date; SELECT COUNT(DISTINCT user_id) FROM tmp_user_daily;这种写法让第一次聚合在小表上进行数据量可以缩小几个数量级。我实测过一个模拟项目一张 30 亿行流水表用户数只有 5000 万先做用户粒度聚合表缩小到 5000 万行COUNT(DISTINCT) 的时间从 2 小时压缩到 20 分钟以内。4.1.1 生成预处理表时也要注意键分布直接GROUP BY user_id, date也会产生 shuffle如果 user_id 的数据分布本身合理没问题如果某些热点用户比如大 V 用户一天的流水特别多就会造成倾斜。解决倾斜的手段前面提过可以加盐或者使用两阶段聚合这个在 4.4 里会具体讲。4.2 方案二使用 ROW_NUMBER 或窗口函数去重有些场景不仅是统计独立数还需要拿到去重后的明细行那么就要用窗口函数去重。常见做法是按需求字段分区再按某个排序字段取第一行WITH ranked AS ( SELECT *, ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY event_time DESC) AS rn FROM raw_table WHERE date 2024-01-01 ) SELECT * FROM ranked WHERE rn 1;这个方案适合“每个用户保留一条记录”的场景比GROUP BY更灵活可以把想要的字段完整保留下来。但缺点也很明显如果分区字段基数极大比如上亿个用户窗口函数依然会在每个分区内全量排序内存压力很大。Spark 对窗口函数的处理是把同一个分区的数据 shuffle 到同一个 executor所以如果某个分区特别大会拖慢整个 job。我碰到过一个案例分区字段是设备 ID总量 20 亿窗口函数跑了一个多小时没结束。最后改成“数据预处理 手动分区排序”的方法才能把时间降到可控范围。所以我的经验是窗口函数适合百万到千万级的分区数量再往上就要考虑预处理了。4.3 方案三用 Spark 内置近似算法这个方案我前面提过了再展开讲讲。approx_count_distinct是一个内置的高性能去重统计方法语法SELECT approx_count_distinct(user_id, 0.02) AS approx_users FROM raw_table;第二个参数是相对误差范围 0~1越小越精确但内存占用越高。在实际使用中0.01 到 0.05 已经是业务可以接受的范围了。HyperLogLog 的核心原理是把输入值哈希成一串比特位通过记录每个桶里的最大前导零数来估算基数不需要保存所有去重后的值内存占用非常小。我拿模拟的 5 亿行数据试过精确 COUNT(DISTINCT) 需要 30 分钟approx_count_distinct只需 45 秒误差率在 1.3% 左右。关键是这个函数在聚合阶段不需要维护完整 key 集合所以 shuffle 的数据量大大减少。4.3.1 极限场景多 distinct 近似如果一条 SQL 里要同时算多个列的近似独立数直接用approx_count_distinct(column, ...)多次即可它支持多个 distinct 参数不会像精确 distinct 那样爆炸。在报表场景里这是一个非常好的替代方案。4.4 方案四处理数据倾斜“数据倾斜”四个字在大数据工程师的字典里大概能排进烦人程度前三。在 DISTINCT 场景中倾斜通常发生在 GROUP BY 去重或先去重后统计的过程中。一个典型的场景需要统计“不同城市独立用户数”一线城市的数据量是大县城的几百倍。GROUP BY city 后某个 reducer 要处理的数据量远超它的兄弟节点整个 stage 就卡住了。解决办法是两个两阶段聚合先给 key 加随机前缀。比如GROUP BY city, rand(50)先把数据打散到 50 个桶里每个桶内部去重然后再对聚合结果按 city 去重。注意这里两阶段都必须用去重逻辑避免二次聚合时重复计数。加盐再聚当某个 key 本身数据量实在太大时可以在key _ rand(100)级别先预聚合再对原 key 聚合。在实际项目中我通常优先使用两阶段聚合代码示例-- 第一层加盐去重 SELECT city, salt, user_id FROM ( SELECT city, user_id, FLOOR(RAND() * 100) AS salt FROM raw_data ) GROUP BY city, salt, user_id; -- 这里按盐值分组去重 -- 第二层按城市聚合去重 SELECT city, COUNT(DISTINCT user_id) FROM salted_dedup_table GROUP BY city;这个方案能显著缓解倾斜但代价是多了一次 shuffle如果倾斜程度不严重可能反倒是负面优化。要先用 Spark UI 看各 task 的数据量分布做到心里有数再决定要不要加盐。4.5 方案五调整动态分区与文件大小如果 DISTINCT 结果要落地成表小文件问题会直接影响后续查询性能。DISTINCT 的 reduce task 数量如果设置过大会同时生成大量小文件。我见过一个任务生成了 3000 多个小文件每个文件不到 1MB后续再来查询就是一场灾难。建议操作是SET spark.sql.adaptive.enabled true; SET spark.sql.adaptive.coalescePartitions.enabled true; SET spark.sql.adaptive.coalescePartitions.minPartitionNum 1;开启 Spark 3.0 的 AQEAdaptive Query Execution之后Spark 会在运行时根据 shuffle 文件大小自动调整 reducer 数量自动合并小文件。这个功能上线后我的很多 DISTINCT 任务都不需要手动调spark.sql.shuffle.partitions了它自己会找到一个合理的分区数。4.5.1 小文件产生的根因其实小文件问题往往不是 DISTINCT 本身造成的而是 reduce task 数量太多或者写到磁盘的分桶策略不对。使用 AQE 能解决大部分“后知后觉”的问题但如果你的 Spark 版本低于 3.0就只能手动在写表时通过distribute by控制文件数量了比如INSERT OVERWRITE TABLE target SELECT DISTINCT ... DISTRIBUTE BY key。5 实操一次 COMPLETE 的把慢查询改成快查询5.1 原始场景还原某个分析任务需求从一张 8 亿行的事实表模拟字段event_id,user_id,device_id,page_id,event_time,channel中计算每日独立用户数以及独立设备数同时要求输出每个渠道的独立用户数。最初的 SQL 是这么写的SELECT date, COUNT(DISTINCT user_id) AS uv, COUNT(DISTINCT device_id) AS device_num FROM fact_table GROUP BY date;加每个渠道独立用户数后SELECT date, channel, COUNT(DISTINCT user_id) AS uv FROM fact_table GROUP BY date, channel;这两个 SQL 单独跑都要花很长时间。我的做法是先做一层“用户-设备-渠道-日期”粒度的预去重再把结果写入一张中间表然后用中间表来统计。5.2 优化后的 SQL 改写第一步抽取最小的去重粒度独立数其实只需要几个字段没必要带上所有的列CREATE TABLE IF NOT EXISTS dws_user_device_daily AS SELECT date, channel, user_id, device_id FROM fact_table WHERE date 2024-01-01 GROUP BY date, channel, user_id, device_id;第二步基于中间表做各种维度的统计SELECT date, COUNT(DISTINCT user_id) AS uv, COUNT(DISTINCT device_id) AS device_num FROM dws_user_device_daily GROUP BY date;以及SELECT date, channel, COUNT(DISTINCT user_id) AS uv FROM dws_user_device_daily GROUP BY date, channel;优化效果非常直观原始扫描表 8 亿行GroupBy 之后中间表大概只有 2 亿行因为同一个用户相同设备在同一天会有多次事件数据量缩小了 75%。第一层聚合虽然也耗时但由于只处理 4 个字段序列化和 shuffle 的开销都降低了整体跑完从 40 分钟缩减到 15 分钟。5.2.1 为什么先把需要去重的字段单独抽出来很多人会问为什么不直接在原始表上 COUNT(DISTINCT)因为原始表里每一行都带着几十个无关字段。Spark 在 shuffle 时需要序列化整个 row字段越多序列化越慢网络传输的数据量越大。抽字段先将 row 瘦身这一步是海量数据下最容易见效的优化。5.3 排查数据倾斜用两段式聚合兜底中间表做出来之后如果某个 channel 记录数特别多比如主站渠道占 80% 数据量按 channel GROUP BY COUNT(DISTINCT user_id) 依然会倾斜。此时我使用了两段式聚合-- 第一段加盐预处理 CREATE TABLE dws_user_device_daily_salt AS SELECT date, channel, FLOOR(RAND() * 100) AS salt, user_id, device_id FROM dws_user_device_daily; -- 第二段按盐值粒度局部去重 CREATE TABLE dws_user_device_daily_salt_dedup AS SELECT date, channel, salt, user_id, device_id FROM dws_user_device_daily_salt GROUP BY date, channel, salt, user_id, device_id; -- 第三段最终统计 SELECT date, channel, COUNT(DISTINCT user_id) AS uv FROM dws_user_device_daily_salt_dedup GROUP BY date, channel;中间多了一个 salt 列数据会在不同 salt 分桶中并行处理最大单个 task 的压力就降下来了。这个方案在倾斜场景下非常有效但也要注意多了两步调度开销非倾斜场景不必强上。6 参数调优中的细节经验6.1 调整 shuffle partitions 的时机spark.sql.shuffle.partitions的默认值是 200很多团队从来没动过。但 200 个分区在几十亿数据量的场景下意味着每个分区要处理上百万条记录单分区工作量大反过来如果数据量只有几百万设成 200 也是浪费很多 task 调度开销。我的经验判断标准很简单看 Spark UI 上每个 task 处理的数据量如果单 task 超过 200MB说明分区数太少调大如果单 task 低于 10MB说明分区数太多调小。针对 DISTINCT 的场景shuffle partition 的作用是控制聚合后的并行度我的常用配置SET spark.sql.shuffle.partitions 800; SET spark.sql.autoBroadcastJoinThreshold 10485760;但如果开了 AQE就不要手动设得太死让框架自己调整更好。这里有一个矛盾点要注意AQE 的coalescePartitions会把过度分区的数据合并但如果你手动把 partition 设成 1000AQE 可能 3 分钟还没开始计算就直接把分区合并到几百反而多了额外开销。所以我建议开 AQE 的情况下spark.sql.shuffle.partitions只需给一个初始值比如 400 或 500剩下的交给动态优化。6.2 开启 Kryo 序列化KryoSerializer通常能比 Java 默认序列化快 5 到 10 倍尤其在大对象和复杂类型聚合时差距明显。配置方法spark.serializer org.apache.spark.serializer.KryoSerializer需要注意使用 Kryo 时如果自定义类需要注册类否则会报错对于标准类型String、Integer、Array则不需要额外注册。我在实践中遇到过一个坑项目的自定义 UDF 返回了一个嵌套 case class忘记注册 Kryo 类导致序列化阶段频繁报错所以如果你要开 Kryo一定要把自定义类型在 conf 里注册好。6.3 内存配置要与 shuffle 策略配合聚合类的操作非常吃内存尤其是 DISTINCT 后接着做 JOIN 的场景。如果 executor 内存不够Spark 会把数据溢出到磁盘性能断崖式下降。我一般建议把spark.memory.fraction保持在 0.6 以上spark.memory.storageFraction设为 0.3 左右给 shuffle 留出足够空间。如果你发现执行计划中频繁出现spill to disk的提示那就是内存不够的直接信号。这时候不要一味加大 executor 内存可能需要优化 SQL 减少一次性加载的数据量比加内存更有效。7 常见问题与排查技巧实录7.1 DISTINCT 总在 shuffle 阶段卡死现象某个 stage 一直显示 99% 但是跑不完看 log 发现某个 task 处理时间远大于其他 task。排查步骤Spark UI 打开 stage 详情查看 task 数据柱状图是否出现极端不均。如果是倾斜对参与 DISTINCT 的 key 进行分布统计比如GROUP BY key看有没有超大 key。用两段式聚合加上盐处理把热点 key 打散。我自己排查的时候经常写一个小查询来看 key 分布SELECT user_id, COUNT(*) AS cnt FROM fact_table WHERE date 2024-01-01 GROUP BY user_id ORDER BY cnt DESC LIMIT 10;能看到排名靠前的 key 的计数就能判断倾斜程度。如果最大 key 计数是平均值的 100 倍以上就必须处理。7.2 多个 COUNT(DISTINCT) 跑得特别慢现象SELECT COUNT(DISTINCT a), COUNT(DISTINCT b), COUNT(DISTINCT c) FROM table跑了半天。优化路径如果能接受近似结果直接换成三个approx_count_distinct。如果必须是精确值先看看这个 SQL 能不能拆成三条单独查询再将结果合并到一张结果表。虽然多跑了几次扫描但每次扫描只需要专注一个列反而比单次多个列更可控。旧版本 Spark 对多 distinct 的优化不友好比如 Spark 2.x 中多 DISTINCT 会导致多次全部扫描。如果版本允许建议升级到 Spark 3.x这版本处理多 DISTINCT 时做了很多优化。7.3 JOIN 条件里有 DISTINCT 结果现象先对一张大表 DISTINCT 得到一个结果集然后 JOIN 另外的表JOIN 时结果集数据不大但很慢。原因DISTINCT 之后如果生成了很多小文件JOIN 读取时就会产生大量文件扫描。解决办法是在 DISTINCT 写入表时控制输出文件数或者打开 AQE 自动合并。7.4 用 Spark UI 判断哪一步最耗时在 Spark UI 的 SQL Tab 下可以看到每个查询的执行片段以及每个节点处理的行数和耗时。我通常先看整体 DAG找出耗时最长的那个 stage再点进去看它的 shuffle read 和 write 量。一个常见陷阱是大家只看“总共花了多久”而不看“数据读写量”。比如一个 stage 显示 20 分钟但它读的数据量是 3TB另一个 stage 显示 10 分钟读数据量只有 200GB前者可能是正常的后者可能有大问题。所以一定要对比数据量来判断。7.5 空值对 DISTINCT 的影响DISTINCT会把 NULL 当作一个合法值也就是说如果你统计COUNT(DISTINCT col)这一列全是 NULL结果会是 1而不是 0。这一点经常在业务报表里引发小到“数对不上”大至“决策错误”的问题。处理方式如果业务上要忽略空值在 DISTINCT 前对列做过滤SELECT COUNT(DISTINCT IF(col IS NOT NULL, col, NULL)) FROM table;或者直接SELECT COUNT(DISTINCT col) FROM table WHERE col IS NOT NULL;我个人更推荐第二种方式因为它在下推阶段就可以剪掉空值数据减少参与聚合的数据量。8 一些值得分享的扩展思路8.1 使用 Bitmap 精确去重如果对精确度和空间占用都有要求可以考虑用位图Bitmap。简单来理解用一个位数组来标记某个用户是否出现出现过的用户对应的位设为 1最后统计位数组里有多少个 1。这个方案在用户数有限的场景下效果极佳5000 万用户只需要几十 MB 内存。Spark 本身不内置 Bitmap但可以通过 UDAF 实现或者使用现成的第三方库。我在实际项目中试过用 RoaringBitmap 实现精确去重对于 5 亿行的场景比 COUNT(DISTINCT) 快好几倍内存占用也可控。缺点是实现复杂度高适合对性能有极致要求且开发资源充足的团队。8.2 使用 BloomFilter 加速存在性判断如果业务上需要判断某个值是否在去重结果集内而不是统计数量可以考虑用 BloomFilter。它能告诉你要找的值“肯定不存在”或“可能存在”有极小的误判率。Spark 的 DataSource API 里可以用追加方式实现 BloomFilter 写入但实际操作中的顺序通常是先在数据写入时生成 BloomFilter 文件然后查询用它做过滤。8.3 预处理思路如何配合实时计算如果数据是流式进入的比如 Kafka 实时主题预处理去重的思路依然适用在流式 ETL 里按窗口聚合做去重定时写出结果表最后在 OLAP 引擎上查询统计结果。Spark Structured Streaming 支持dropDuplicates操作指定主键字段去重效果等同于 DISTINCT但基于状态存储实现能增量处理。这种方案的优点是查询侧完全没有压力缺点是流式状态后台存储会随着 key 的规模增长而膨胀需要配合过期清理机制。我在模拟实践里用事件时间窗口做滚动去重效果比离线 ETL 更加及时。9 最后补充一些实操心得个人在实际项目里踩过不少坑之后最想强调的是DISTINCT 优化不是单一手段能解决的而是组合拳。首先判断业务到底需要精确还是近似再看数据分布是否需要加盐处理然后决定是改写 SQL 还是调整配置参数。现在很多优化方向其实是把压力前移用预处理换查询性能这不丢人反而是一种工程上的正确选择。如果你当前的任务只是零星地算几次独立数花时间优化可能不值得但如果这是一个每天都要跑的报表任务强烈建议花半天时间把预处理的链路搭好把中间表建出来长期来看收益非常可观。还有一个小技巧写 DISTINCT 查询时尽量把要参与去重的字段放在 SELECT 的最前面Spark 优化器在列剪枝时可能会因此少处理一些列。这个说法我没有严格验证过但从实际操作看保持查询列精简总归没有坏处。
