hive 中 group by 和 distinct 孰优孰劣？

hive 中 group by 和 distinct 孰优孰劣？

首先声明一下，hive是什么：

hive 不是数据库，hive 只是一个数据仓库工具，可以用来查询、转化和加载数据，是可以调用 mapreduce 任务、用类 mysql 语法查询HDFS数据的一个工具。 再来说 mapreduce 是什么，mapreduce 是分而治之的一种编程模型，适用于大规模数据集的并行计算，当处理一个查询任务时，先调用 map 任务并行处理，最后用 reduce 任务归约结果。

那么对于一张200w+的表，我们来看一下group by 和 distinct 的执行过程：

set mapreduce.map.java.opts="-Dfile.encoding=UTF-8"; set mapreduce.reduce.java.opts="-Dfile.encoding=UTF-8";
select province,city from seven_dataset_67 group by province,city

set mapreduce.map.java.opts="-Dfile.encoding=UTF-8"; set mapreduce.reduce.java.opts="-Dfile.encoding=UTF-8";
select distinct province,city from seven_dataset_67 

这里可能就要问了，不是通常说 group by 的效率高于distinct吗？为什么distinct 和 group by 都调用了 18 个 reduce 任务，distinct 的时间还花费的少一些，其实在数据量不大的情况下，distinct 和 group by的差别不是很大。但是对于count(distinct *) 来说会发生数据倾斜，因为 hive 默认在处理COUNT这种“全聚合(full aggregates)”计算时，它会忽略用户指定的Reduce Task数，而强制使用 1，会发生数据倾斜。

select count(0) from seven_dataset_67 

但是仍然要看数据量的大小，在数据量小的情况下去重计数，虽然count(distinct *) 会发生数据倾斜，但是只有执行一次 mapreduce任务，而 select count(0) from(select field from table group by field) 这种要执行两遍 mapreduce 任务，总的时间花费可能不比前者少，如下例子：

select count(0) from (select sales_order_no from seven_dataset_67 group by sales_order_no) a

*但是针对上亿的数据量，数据倾斜就会浪费很多时间，甚至由于机器资源紧张导致运行失败，这种情况就建议使用group by了，不仅可以分组，还能配合聚合函数一起使用*

小结

1. 基础定位差异

   • GROUP BY：是分组聚合操作，可以配合聚合函数使用，天然具备去重功能

   • DISTINCT：是纯粹去重操作，语法更简洁但功能单一

2. 性能对比关键发现

   场景	GROUP BY优势	DISTINCT优势
   大数据量复杂分析	多Reducer并行处理，避免单点压力	-
   简单去重小数据集	-	执行计划更简洁，减少中间步骤
   多字段去重	支持多字段灵活组合	语法更直观
   数据倾斜场景	可通过hive.groupby.skewindata优化	强制单Reducer处理，风险高

3. 执行原理差异

   • GROUP BY：Map阶段按分组字段生成Key，Reduce阶段聚合，支持多Reducer并行
   • DISTINCT：Map阶段输出全字段作为Key，强制单Reducer去重

实际场景选择建议

对于mysql数据库来说，哪个性能更好？

在 Mysql8.0 之前 group by 会进行隐式排序，导致触发 filesort，sql 执行效率低下， distinct 效率高于 group by。但从 Mysql8.0 开始，Mysql 就删除了隐式排序，所以在语义相同，无索引的情况下，group by 和 distinct 的执行效率也是近乎等价的。

为什么更推崇使用group by？

1. group by 语义更为清晰
2. group by 可对数据进行更为复杂的一些处理，相比于distinct来说，group by 的语义明确。且由于distinct 关键字会对所有字段生效，在进行复合业务处理时，group by 的使用灵活性更高，group by 能根据分组情况，对数据进行更为复杂的处理，例如通过 having 对数据进行过滤，或通过聚合函数对数据进行运算

数据倾斜及优化

数据倾斜：即数据分布不均匀导致某些reduce处理数据量过大

数据倾斜实现原理

join实现原理

select name, orderid
from user t1
join order t2
on t1.uid=t2.uid

group by 实现原理

select rank, isonline, count(1)
from city
group by 1, 2

原因

• key分布不均匀
• 业务数据本身的特性
• 建表时考虑不周
• 某些SQL语句本身就有数据倾斜

具体表现

任务进度长时间维持在99%（或100%），查看任务监控页面，发现只有少量（1个或几个）reduce子任务未完成。因为其处理的数据量和其他reduce差异过大。

单一reduce的记录数与平均记录数差异过大，通常可能达到3倍甚至更多。 最长时长远大于平均时长。

例如：看到下面这种情况，这肯定是数据倾斜了。map早就完工了，reduce阶段一直卡在99%，而且cumulative cpu的时间还一直在增长，说明整个job还在后台跑着。这种情况下，99%的可能性就是数据发生了倾斜，整个查询任务都在等某个节点完成。

如何解决

Hive的执行是分阶段的，map处理数据量的差异取决于上一个stage的reduce输出，所以如何将数据均匀的分配到各个reduce中，就是解决数据倾斜的根本所在。

join引起数据倾斜的解决方法

关联字段中有空值或者脏数据

如果是由于key值为空或为异常记录，且这些记录不能被过滤掉的情况下：可以考虑给key赋一个随机值，将这些值分散到不同的reduce进行处理。由于null值关联不上，处理后并不影响最终结果。

关联字段中key都为有效值

关联字段中key都为有效值，某些key量大，造成reduce计算量大

• 如果是一个大表和一个小表join的话，可以考虑使用mapjoin来避免数据倾斜，mapjoin的具体过程如下。分为两步：

1. 通过mapreduce local task, 扫描小表，生成为一个hashtable文件, 并上传到distributed cache

2. 在map阶段，每个mapper, 从distributed cache中读取hashtable文件，扫描大表，并直接在map端join

• 设置reduce个数参数，提高reduce个数：

set hive.exec.reducers.bytes.per.reducer = 1000000000或
set mapred.reduce.tasks=800 这两个一般不同时使用，

• 设置每个reduce处理的数据量的大小：

set hive.optimize.skewjoin = true;

set hive.skewjoin.key = skew_key_threshold （default = 100000）
set hive.skewjoin.key = 250000000

可以就按官方默认的1个reduce 只处理1G 的算法，那么skew_key_threshold= 1G/平均行长；或者默认直接设成250000000 (差不多算平均行长4个字节)

• 巧用MapJoin解决数据倾斜的问题(小表join大表时)

Hive的MapJoin理解：join的操作是在map阶段完成后，如果需要的数据在map的过程中可以访问到则就不再需要reduce了。

例如：小表关联一个超大表时，容易发生数据倾斜，可以使用Mapjoin把小表全部加载到内存，广播的方式分发到不同的map中，在map端进行join，避免reduce处理

select c.channel_name,count(t.requesturl) PV  
from ods.cms_channel c  
join  
(select host,requesturl from  dms.tracklog_5min where day='20241111' ) t  
on c.channel_name=t.host  
group by c.channel_name  
order by c.channel_name;  

上面的是一个小表join一个大表的时候，可以使用mapjoin把小表放到内存中处理，语法只需要增加 /*+MAPJOIN(表的名字)*/

select /*+ MAPJOIN(c) */  
c.channel_name,count(t.requesturl) PV  
from ods.cms_channel c  
join  
(select host,requesturl from  dms.tracklog_5min where day='20241111' ) t  
on c.channel_name=t.host  
group by c.channel_name  
order by c.channel_name;

数据倾斜的时候，常常如上面这么使用

一般认为在25M以内的数据都是小表：hive.mapjoin.smalltable.filesize=25000000

group by 引起数据倾斜的解决方法

group by 引起数据倾斜的原因是 group by 维度过小，某值的数量过多

• 设置在map端进行combiner聚合，

set hive.map.aggr=true

开启map之后使用combiner，但是这个通常对数据比较同质的有用，相反，则没有什么意义。

• 通用的做法是设置下面两个参数：

set hive.groupby.mapaggr.checkinterval = 100000 (默认)执行聚合的条数

set hive.map.aggr.hash.min.reduction=0.5(默认)如果hash表的容量与输入行数之比超过这个数，那么map端的hash聚合将被关闭，默认是0.5，设置为1可以保证hash聚合永不被关闭；

还有一个是set hive.groupby.skewindata=true， 这个只针对单列有效。

• HiveQL中包含count（distinct）时

如果数据量非常大，执行如 select a,count(distinct b) from t group by a; 类型的SQL时，会出现数据倾斜的问题。

解决方法：采用sum() group by的方式来替换count(distinct)完成计算。

select a,sum(1) from (select a, b from t group by a,b) group by a;