一、数据倾斜概念
数据倾斜(Data Skew)问题是指在分布式计算系统中,由于某些数据分布不均匀,导致某些节点处理的数据量远远大于其他节点,从而引起性能瓶颈、计算延迟甚至计算失败的现象。
在Hive中,在处理分组聚合(Group By)或者多表连接(Join)问题时,如果分组字段或者连接字段存在分布不均匀,即某个key或者某些key的数量远超其他key,就会造成在按key分区发往reduce端的时候,某个reduce接收了大量的数据,所需的处理时间也会远超其他reduce,成为整个任务的瓶颈。
二、Hive解决数据倾斜——分组聚合导致
常规情况下,hive的分组聚合是由map端读取数据,按照分组字段key进行分区,并通过shuffle发往reduce端进行聚合。如果分组字段分布不均匀,就会导致大量相同的key发往同一个reduce,从而导致数据倾斜问题。由分组聚合导致的数据倾斜问题,主要有以下两种解决思路(原理都是先完成部分聚合再发往reduce端):
(1)map端聚合(Map-Side聚合)
开启map端聚合后,数据先在每个并行的map端分别进行部分聚合,聚合之后再发往reduce端,此时可以在很大程度上解决数据倾斜问题。主要涉及以下四个参数:
# 开启map端聚合
set hive.map.aggr=true;# 判断是否使用map端聚合,因为可能字段的差异性很大,聚合后数据不会减少多少,即map端聚合意义不大。 具体来说,对源表前100000条数据进行聚合检测,如果聚合后的数据条数/100000 < 0.5则使用map端聚合。
set hive.map.aggr.hash.min.reduction=0.5;
set hive.groupby.mapaggr.checkinterval=100000;# map端聚合所使用的hash table占用map task堆内存的最大比例,超过即flush
set hive.aggr.hash.force.flush.memory.threshold=0.9;
(2)Skew-GroupBy优化
Skew-GroupBy优化的原理是:开启两个串行执行的MR任务,第一个MR任务按照随机数进行分区发往reduce端,在reduce完成部分聚合;第二个MR任务再根据分组字段进行分区,完成最终聚合。相关参数如下:
# 启用分组聚合数据倾斜优化
set hive.groupby.skewindata=true;
三、Hive解决数据倾斜——Join导致
若不开启任何优化,hive进行join操作默认是common join算法,即map端负责读取数据,参与join的两张表根据关联字段进行分区,发往reduce端,那么相同的key会发往一个reduce端(两张表),在reduce端完成最后的join操作。(类似mapreduce中的reduce join)
如果分组字段key分布不均匀,那么某个reduce端需要处理的数据就远远超过其他reduce,造成数据倾斜,发生性能瓶颈。对于join导致的数据倾斜问题,有如下解决思路:
(1)map join
map join的前提是大表join小表的场景,使用map join则没有reduce阶段,则没有shuffle,自然不会产生数据倾斜问题。map join算法由两个只有map阶段的MR任务完成一个join操作, 其本质是第一个MR任务读取满足条件的小表的数据,将其制作为hashtable,并上传至分布式缓存中,第二个MR任务会先从缓存中读取小表数据,并缓存在自己的map task缓存中,扫描大表进行join操作。涉及到的参数如下:
# 开启map join自动转换
set hive.auto.convert.join=true;# 不开启无条件转mapjoin的小表的判定标准
set hive.mapjoin.smalltable.filesize=250000;# 开启无条件转mapjoin
set hive.auto.convert.join.noconditionaltask=true;
# 开启无条件转mapjoin的小表判定标准
set hive.auto.convert.join.noconditionaltask.size=10000000;
(2)skew join
map join需要一个表是小表,对于大表join大表的场景不适用(思考此时用bucket map join 和SMB map join有用吗?)。
skew join的原理是,为倾斜的大key单独启动一个map join任务进行计算(实际生产环境中的join大多是一对多,大key是多的一方作为大表,另外一边是一,可以作为小表),其余key进行正常的common join。
# 开启skew join 优化
set hive.optimize.skewjoin=true;
# 当某个key的数量超过100000,则认为存在数据分布不均,触发skew join(即触发阈值)
set hive.skew.join.key=100000;
(3)使用 Salting(加盐技术)
加盐 是指在某些字段上加上随机值,打散倾斜的键。例如,当某个字段的值分布不均匀时,可以在字段值上加上一些随机数,让数据在 Shuffle 过程中更加均匀地分配到不同的 Reducer 上。
如以下,假设id是倾斜字段,则通过concat随机数,将id字段同一个值打散为两个值,原本相同的key会被发往两个reduce端进行join操作。
select*
from(select --打散操作concat(id,'_',cast(rand()*2 as int)) id,valuefrom A
)ta
join(select --扩容操作concat(id,'_',0) id,valuefrom Bunion allselectconcat(id,'_',1) id,valuefrom B
)tb
on ta.id=tb.id;