Hive去重优化：distinct与group by性能对比及调优实战

做数据的人，大概都写过这类SQL：统计今日UV、统计独立用户数、统计某维度组合的去重数量。distinct和group by在SQL语义上经常可以互相替代，所以网上总有人争论哪个更快。我早年也以为这俩差不多，直到有一次线上一个统计任务用distinct跑了快两个小时还卡在最后的Reduce阶段，换成group by写法半小时就出结果了。从那以后，我对这个问题的态度就变成了：看场景、看引擎、看数据分布，没有绝对的好，但有明确的取舍逻辑。

这篇文章我会从Hive的执行原理讲起，再用实际场景对比两者的性能差异，接着给出一套可落地的调优方案，最后聊聊Hive 3.x、Tez/Spark这些新引擎下局面发生了什么变化。不管你是刚接触Hive的新人，还是已经在生产环境里踩过坑的老手，希望能帮你把group by和distinct的选择逻辑彻底理顺。

1. 先搞懂执行原理：distinct和group by在Hive里到底怎么跑

1.1 从MapReduce模型看两者的本质

要判断谁快谁慢，得先知道Hive底层怎么执行这两类SQL。在Hive 2.x及更早版本，默认执行引擎是MapReduce，整个计算过程分为Map、Shuffle、Reduce三个阶段。

distinct本质上是全局去重。在Map阶段，Hive会把去重字段作为key输出，但Shuffle之后，为了确保“同一字段值只能出现一次”，数据必须汇到同一个Reduce任务上做全局去重。换句话说，经典执行计划下，select distinct user_id from t这条SQL无论数据量多大，Reducer数量基本就是1个。单Reduce意味着所有Map输出的数据全部压向一台机器，网络IO、磁盘读写、单节点计算压力全部集中在这一个点上，这为数据倾斜埋下了很大的隐患。

group by则不一样。它天然支持分治：Shuffle阶段按分组key进行分区，每个分组key会路由到同一个Reducer，但多个分组可以并行分布在不同Reducer上。更关键的是，Hive默认开启hive.map.aggr=true，Map阶段会对相同key先做一次局部聚合，也就是Map端聚合。这个过程会把大量重复数据在Map端就提前削减掉，Shuffle的数据量往往是原始数据的一个零头。

我用一个生活化的类比来帮你理解。distinct就像全班同学把作业统一交到班主任一个人手里，班主任在讲台上一份一份检查有没有重复的名字，全校几千人的作业只由一个人处理；group by则是先按班级分组，各班班长在自己教室先整理一遍名单，只把“班级+唯一姓名”汇总到年级组，年级组再合并，压力分散、流量减少，自然更快。

1.2 必须正视的“单Reduce魔咒”与数据倾斜

上面说的“单Reduce”是Hive早期版本中distinct最致命的问题。实际生产环境里，数据分布几乎不可能是均匀的。比如统计按来源渠道分组后的去重用户数，某些大渠道的用户量可能是小渠道的几百倍。使用distinct时，所有渠道的数据全都进入同一个Reducer，热点key直接被放到一台机器上处理，OOM、任务失败、长时间卡顿都很常见。

即使改用group by source_channel，如果某个渠道的key特别大，也会出现单个Reducer倾斜。但区别在于，group by可以对倾斜的key进行加盐拆分、Salted Shuffle等处理后，把大key打散成多份并行处理，而distinct因为是全量全局去重，加盐处理的逻辑必须额外小心，不能直接套用，否则去重结果会出错。

还有一个很容易被忽视的点：distinct在Shuffle阶段没有Map端聚合可以依赖。Map端聚合依赖“相同key在本地就有足够多的重复”这个前提，distinct虽然也是按字段分组，但map端做局部去重后，如果原数据中某个字段值出现次数只有1次，局部去重几乎没有削减效果，全部数据照样原封不动发给单个Reducer。

1.3 group by的Map端聚合优势到底有多大

group by的Map端局部聚合在很多场景下效果惊人。假设你有一个10亿条的日志表，要统计每天的UV，即select day, count(distinct user_id) from log group by day。先别管这段SQL里的distinct，单看group by day，Map阶段每个MapTask会先在本地算出一份(day, 部分聚合结果)，然后只需要把少量Map端结果Shuffle给Reduce。

我在实际测试里遇到过这样的情况：一张日增量5亿条的曝光日志，按用户维度用group by做预聚合，Shuffle的数据量从原始的上百GB直接降到几GB，原因是单个用户一天的曝光记录可能有几十上百条，Map端聚合能把重复记录拍掉一大半。后续Reduce阶段的压力也成倍下降，整个任务的耗时自然大幅缩短。

但注意，group by的Map端聚合不是万能的。如果分组字段基数特别大，比如直接用毫秒时间戳或者完整URL做分组维度，Map端聚合基本起不到削减作用。另外，如果聚合函数本身依赖全局状态，Map端聚合也可能失效，这类问题牵扯到UDAF和Combiner的关联，我后面会单独展开。

2. 实测对比：不同场景下谁更划算

2.1 单列去重场景：group by真的稳赢吗

我做了很多次对比测试，场景尽量贴近生产：一张模拟电商订单表，字段包括order_id, user_id, order_date, amount, channel。测试SQL非常简单，一个算总去重用户数，一个算按渠道分组的去重用户数。

单列去重总量这个场景，我拿1亿条订单数据测，select count(distinct user_id)和select count(*) from (select user_id from t group by user_id) tmp，前者默认落在一个Reducer上，跑完大概需要18分钟；后者因为Map端聚合把重复的user_id压掉了一大批，Shuffle数据量小了很多，最终耗时大概12分钟。两个SQL如果加上order_date做为时间过滤条件，数据量降到几千万条时，差异会缩小到一两分钟以内。

这意味着什么？如果数据量不大，或者去重字段的基数远小于总记录数，group by的优势没有被完全激发，两者耗时的差距就没那么明显，甚至可以互相替换。但一旦数据量上亿，且去重字段有大量重复值，group by的Map端聚合优势会被放大，差距能到30%以上。

所以我的结论不是“无脑用group by”，而是“在数据量大、重复率高的场景，group by更稳；在数据量小、字段基数又接近总行数的场景，两者基本差不多”。

2.2 多列去重与多个count(distinct)的经典组合

现实业务里，只去重一列的情况太少见了。更常见的写法是：

select channel, count(distinct user_id) as uv, count(distinct order_id) as order_cnt, count(distinct product_id) as product_cnt from orders where day = '2025-01-01' group by channel;

这类SQL有三个count(distinct)，在Hive 2.x的MapReduce引擎下，会触发多次MapReduce Job。原因很简单：Hive的优化器在应对多个不同字段的distinct时，很难在单个Reduce阶段同时完成三种不同的全局去重。最终表现就是Job数量翻倍，每一轮都要全量扫描数据，整个任务耗时甚至比三个单独SQL加起来还要夸张。

另外，还需要区分count(distinct a, b)和count(distinct a), count(distinct b)。前者是对(a, b)组合去重，后者是分别对a和b去重，两者统计口径完全不同，但在执行计划里都叫“Distinct”。组合去重在部分Hive版本中同样只能走单Reduce，性能更差。

针对这种多列去重场景，我更推荐先group by去重，再做外层聚合：

select channel, count(*) as uv, count(order_id) as order_cnt, count(product_id) as product_cnt from ( select channel, user_id, order_id, product_id from orders where day = '2025-01-01' group by channel, user_id, order_id, product_id ) t group by channel;

内层group by把多列一起分组，天然形成了“组合去重”的中间结果；外层再按渠道做纯count(*)聚合，每一步都不涉及全局单Reduce，执行计划更均衡，Job数量也稳定。

2.3 group by + 再聚合的组合拳能解决什么

“先group by去重，再二次聚合”这招，几乎可以覆盖所有去重统计口径，而且正确性非常好理解。它尤其适用于三类场景。

第一类是多列去重计数，上面已经演示过了。第二类是不同时间窗口的UV计算，比如要同时看当日UV、7日UV、30日UV。如果只用count(distinct user_id)加case when，写出来的SQL不仅冗长，性能还差。更好的方式是先按user_id聚合出活跃日期列表，然后在外层按时间窗口条件计数。第三类是留存分析里的多日用户去重，比如同时看昨天和今天的活跃用户交集，用group by结合多个sum(case when ...)可以清晰表达。

我用一个留存口径的例子来说明：

select a.day, count(distinct case when b.day = '2025-01-02' then a.user_id end) as active_retain_uv from ( select user_id, day from user_active_daily where day = '2025-01-01' group by user_id, day ) a left join ( select user_id, day from user_active_daily where day = '2025-01-02' group by user_id, day ) b on a.user_id = b.user_id group by a.day;

这个SQL先用双层group by把两天的活跃明细各自去重，再通过join关联，最外层计数。count(distinct case when ...)在这里只在结果集很小的外层使用，内层完全不碰全局distinct，整个执行计划不容易出倾斜问题。

3. 调优实战：数据倾斜、小文件与UDAF

3.1 数据倾斜的经典解法：加盐与两阶段聚合

数据倾斜是group by和distinct都会遇到的问题。distinct的倾斜体现在单Reduce压垮一台机器，group by的倾斜体现在某个大key霸占一个Reducer。针对group by倾斜，最稳妥的方案是两阶段聚合。

第一步，给key加一个随机扰动，比如concat(user_id, '_', floor(rand() * 10))，让原来的大key被拆成10份；第二步，在每个小分组内做局部聚合；第三步，去掉盐分再按原key聚合一次。这个过程其实对应了很多数据计算引擎中的“Partial Aggregation + Final Aggregation”。

用SQL写出来大致是：

select channel, sum(partial_cnt) as uv from ( select channel, concat(user_id, '_', floor(rand() * 10)) as salted_key, count(*) as partial_cnt from orders group by channel, concat(user_id, '_', floor(rand() * 10)) ) t group by channel;

注意，这段SQL处理的是“用户出现多次只算一次”的UV口径。加盐后，同一个用户会被随机拆到10个不同的小桶里，每个小桶里的count(*)都只是该用户被分到这个桶的记录数。外层把partial_cnt全部加起来，恰好等于该渠道下所有用户的出现次数和。如果每个用户只出现一次，算出来的就是真正的UV；如果用户有重复出现，这个方案就不适用了，需要先对(channel, salted_key, user_id)做去重再计数。

不过Hive本身的count(distinct user_id)更习惯于直接使用：

select channel, count(distinct user_id) from orders group by channel;

这种写法里distinct和group by共存，Map端可以把大部分重复值先处理掉，Reduce端再对每个channel做去重。如果channel本身倾斜严重，再考虑加盐方案，但加盐去重比较麻烦，需要确保同一个用户始终进入同一个盐分桶，比如改成concat(user_id, '_', hash(user_id) % 10)，不能完全随机。

Hive还提供了一个现成参数hive.groupby.skewindata=true，开启后Hive会尝试把倾斜的group by拆成两个MapReduce Job：第一个Job对加盐后的key做部分聚合，第二个Job对结果做最终聚合。这个参数在常规业务场景已经能解决大部分group by倾斜，但代价是额外多一轮计算，任务时间可能变长，所以不要无脑开启，只在确认倾斜时使用。

3.2 小文件问题与group by的关联治理

很多团队在做Hive优化时，痛点不是SQL性能本身，而是小文件问题。有段时间我排查一个离线数仓任务，发现输出目录下有上千个几十KB的小文件，每次下游加载都慢得离谱，后来定位到根因就是group by的Reduce数量太多，而且每个Reducer的数据输入量差异巨大，输出被切割得很碎。

Reduce数量主要由hive.exec.reducers.bytes.per.reducer控制，默认是256MB或者1GB，具体看版本。在实际调优中，我会按这个公式估算：

reducer数量 = min(hive.exec.reducers.max, 输入数据量 / hive.exec.reducers.bytes.per.reducer)

如果每个Reducer处理的数据量远小于阈值，比如只有几十MB，Reduce数量就会很多，输出的小文件自然也多。解决办法是调大hive.exec.reducers.bytes.per.reducer，让每个Reducer分配到更多数据，减少Reduce数量。但要注意，Reduce数量太少也不行，比如一个Reduce扛10GB数据，单个任务会慢到怀疑人生。

另一个治理方式是强制合并小文件。hive.merge.mapfiles=true和hive.merge.mapredfiles=true可以分别控制Map Only任务和MapReduce任务的输出合并。还有hive.merge.smallfiles.avgsize，可以设置一个平均大小阈值，小于这个阈值就触发合并。这些参数在动态分区插入场景尤为重要，因为动态分区本身就容易产生大量小文件，group by只是加剧了问题。

我遇到过最典型的组合坑是：一张大表动态分区按天写入，分区键的基数虽有1000多个，但每个分区数据量很小，结果写出来的文件全是几十KB的小块。我当时的处理方式是先把原SQL拆成“先聚合中间结果，再动态分区写入”两步，并且给中间结果的外层查询单独设置更合理的Reduce数量，再配合distribute by和文件合并参数，最终把小文件数量降了一个数量级。distribute by这里的作用是控制数据如何分布给Reducer，比如distribute by day可以保证同一天的数据进同一个Reducer，每个Reducer输出一个针对该分区的文件，避免每个Reducer都朝所有分区写碎片。

3.3 为什么count(distinct)不能用Combiner，以及自定义UDAF的用武之地

Hive的Map端聚合，底层依赖一个能力叫Combiner。Combiner对聚合函数有一个隐藏要求：函数必须是“结合律”的，中间结果可以任意合并而不影响最终结果。count(distinct x)不满足这一要求，因为distinct要求全局去重，Map端把部分数据合并成“部分去重后的集合”后，这个集合之间再做合并时，还需要感知全局的重复情况，这个过程没法用简单的计数规约来完成。所以count(distinct)在很多版本里不能走Combiner优化，部分数据照样全量穿透到Reduce。

这个限制也给自定义UDAF留出了空间。比如Hive提供了approx_count_distinct这个函数，底层基于HyperLogLog算法，不精确但速度快。它允许在Map端记录一个固定大小的HLL位图，Combine阶段直接合并位图，Reduce阶段估算基数。实测下来，在百亿级数据上做近似去重，approx_count_distinct比count(distinct)快非常多，误差通常能控制在1%以内，很多大厂做UV实时统计用的就是这类方案。

如果业务要求精确去重，又不想忍受count(distinct)的单Reduce问题，一般会自己写一个UDAF。例如实现一个基于Roaring Bitmap的UDAF，把字段值映射成整数并压入Bitmap，Map端局部构造Bitmap，Combine阶段合并Bitmap，Reduce阶段统计BitMap的基数。我见过不少团队把这种方案用在数亿规模的ID去重上，效果比直接count(distinct)稳定很多。

写自定义UDAF时有几个细节要特别注意。第一，initialize、iterate、merge、terminatePartial、terminate五个方法必须完整实现，尤其不能漏掉terminatePartial，没有它Map端聚合不会生效。第二，如果UDAF需要维护大量中间状态，比如Bitmap占内存很大，要考虑在Map端控制单个key的状态大小，必要时可以结合加盐进一步拆分。第三，UDAF的返回类型如果是复杂结构，要记得实现resolve方法让Hive能正确解析类型。

4. 新引擎与新技术：Hive 3.x、Tez/Spark与Doris的对比

4.1 Hive 3.x + Tez/Spark下，局面发生了哪些变化

很多朋友的印象还停留在“Hive用MapReduce，distinct一定慢”。Hive 3.x以后，事情有了变化。执行引擎换成Tez或Spark后，distinct不再只能走单Reduce。Spark SQL里的distinct会先做一次局部去重，再做一次全局去重，类似于两阶段聚合。Tez引擎下，Hive优化器也会尝试对去重查询做更多改写，尽量分散压力。

我分别在CDH 6.x的Hive 2.x + MapReduce环境和一个Hive 3.x + Tez环境上跑同样的SQL：10亿条记录，统计唯一order_id数量。Hive 3.x + Tez环境下，count(distinct order_id)和“内层group by order_id、外层count(*)”的耗时差距已经缩小到百分之十几，不像MapReduce时代动辄翻倍。但即使如此，group by在重复率高的大数据集上依然更稳，因为它的Map端聚合优势依然存在。

Hive 3.x还引入了LLAP这个概念，全称是Live Long and Process。LLAP让一部分常驻Executor留在节点上，缓存热数据，执行Short-lived查询时不需要重新启动Container。这个能力对交互式单条查询比如select count(distinct user_id) from table where condition帮助很大，因为瓶颈从“容器启动+调度开销”转移到了实际计算过程。但LLAP对内存要求高，集群压力原本已经很大时，开启LLAP反而可能把节点整崩，需要谨慎评估。

4.2 当你手边同时有Hive和Doris，SQL去重该交给谁

相关热搜词里有个hive与doris，这个组合其实代表了批处理和MPP数据库的分界问题。Doris是一个MPP架构的OLAP数据库，数据以列存为主，查询走向量化执行，MPP天然支持并行shuffle，去重、分组、join这些操作都分散到多个BE节点同时执行。在同等数据量下，Doris执行select count(distinct user_id)这类去重查询，速度往往比Hive快出一个量级，因为它不需要启动一堆Container，节点本身常驻，内存表结构也在。

但Doris能替代Hive吗？不能。Hive的价值在于海量离线数据的批处理、复杂的ETL流程、与HDFS生态的深度绑定，以及数据分布在超大集群时的稳定性。Doris更适合数据已经加工好之后，提供给业务方做交互式分析和报表查询。

所以我的建议非常明确：如果瓶颈出现在Hive的离线加工链路里，比如每天凌晨跑数，耗时在小时级别，那么优先优化Hive本身的SQL、参数和UDAF；如果需要给业务方提供一个秒级查询的入口，让他们自己对UV、去重用户做灵活分析，那把数据同步到Doris这类MPP仓库，让Hive做批处理，让Doris做查询，分工明确。

实际项目里，我见过太多团队试图用Hive扛交互式查询，最后把离线集群资源吃干榨净，业务方还是嫌慢。反过来，也有人试图用Doris跑几百亿规模的全量历史数据加工，结果导入超时、内存爆炸。正确思路是画清楚两者的边界：Hive负责“算出来”，Doris负责“查得快”。

5. 常见问题与排查实录

5.1 一句话排掉“列不存在”“group by having错误”这些坑

关于去重和分组相关的报错，我收集了几个高频问题，列成表方便你排查：

这里特别说一下group by和having的配合。having是对分组后的结果进行过滤，所以having里引用的字段只能是分组字段或聚合函数的结果，不能引用原始明细字段。比如select day, count(distinct user_id) as uv from log group by day having uv > 100，这条在多数Hive版本会报错，因为having里不能直接用别名uv，要写成having count(distinct user_id) > 100。习惯了这个规则，写完SQL先自查一遍，能省掉很多无效调试。

5.2 一个真实案例：按日去重用户数的排查全过程

最后分享一个我记忆很深的真实问题。某个业务需要统计最近7天每天的去重用户数，SQL大概长这样：

select day, count(distinct user_id) as uv from user_action_log where day >= date_sub(current_date, 7) group by day;

线上集群是Hive 2.x + MapReduce。第一次跑，任务在Reduce阶段卡了将近40分钟，然后报OOM。我当时先看了两个东西：一是Reduce数量，结果日志显示只有1个Reduce，符合count(distinct)的单Reduce特征；二是数据分布，发现最后一天的数据量占了7天总量的70%左右，全部压向一个Reduce。

我给的优化分两步。第一步，先把瓶颈切掉，改用内层group by day, user_id去重，外层再group by day计数，这样Reduce可以按天分布，理论上7个Reducer并行，每个Reducer只处理一部分数据。

select day, count(*) as uv from ( select day, user_id from user_action_log where day >= date_sub(current_date, 7) group by day, user_id ) t group by day;

第二步是对这个方案的数据倾斜做预防。因为最后一天数据量异常大，仅仅按天分组还不够，大概率出现某一天的单Reducer还是吃不消。我就在内层group by时给user_id加了一个盐值，比如substr(hash(user_id), 1, 3)，拆成更多局部桶，外层再按天汇总，相当于绕开了单天倾斜。

为了验证SQL改动后结果是否正确，我把旧SQL和新SQL各抽了100万条数据做测试对比，两个结果的UV完全一致。正式环境跑下来，整个任务从40分钟降到15分钟左右。后来我又把引擎切到Tez，时间进一步压缩到七八分钟。这个案例其实就把前面讲的原理都串联起来了：单Reduce带来的压力、group by的并行优势、加盐处理大key、新引擎对性能的改变。

5.3 关于hive引擎升级后SQL兼容性的一点点提醒

切引擎或者升级Hive大版本时，distinct和group by的语义基本不变，但执行计划和资源使用会变化。比如Hive 2.x里跑得好好的SQL，切到Spark后可能因为Spark默认的并行度、Shuffle分区数不同而变慢；切到Tez后，部分场景下Tez的容器复用机制会让任务启动变快，但如果你之前手动调了非常细致的Reduce参数，新引擎有可能不按你的预期执行。升级前一定要做一轮全量SQL回归验证，重点对比“数据量接近生产级的测试表上，去重结果是否一致、耗时是否正常”，不要只看一两条SQL。

写自定义UDAF或依赖某个UDAF时，同样要做引擎兼容性验证。Hive 3.x对UDAF的接口有了更严格的检查，老版本里能跑的UDAF在新的GenericUDAFResolver2体系下可能直接加载失败，这类坑在跨版本升级时防不胜防。

写在最后的一个选择建议

我自己在做技术选型时的判断标准已经比较固定了：数据量不大、去重字段又有索引类条件过滤时，怎么写都行，按业务可读性来；数据量大且字段重复率高，优先group by；多列去重统计，优先“内层group by + 外层聚合”；命中数据倾斜，加盐两阶段聚合；追求极速且可以接受近似值，直接上approx_count_distinct或自研HLL/Bitmap类UDAF；集群已经是Hive 3.x或Spark引擎，distinct的劣势缩小了，但仍要多观察执行计划，而不是听网上某一个版本的结论用到死。

如果非要说一条最有价值的经验，那就是：不要在写SQL之前就争论孰优孰劣，先看一眼执行计划，再拿真实数据量跑一次对比，最后根据Reduce数量和Shuffle量级做决定。数据工程师的直觉很重要，但生产环境的数据分布才是最终裁判。