数据倾斜的原因及解决方案

一、什么是数据倾斜及数据倾斜是如何产生的

数据倾斜的根本原因是数据的key分布不均，个别key数据很多，超出了计算节点的计算能力的结果；

过程：数据经过 map后，由于不同key 的数据量分布不均，在shuffle 阶段中通过 partition 将相同的 key 的数据打上发往同一个 reducer 的标记，然后开始 spill （溢写）写入磁盘，最后merge成最终map阶段输出文件。

如此一来 数据量很大的key 将发往同一个 reducer，超出了节点的计算能力，等待时间超出了可接受范围。

二、怎么判断数据有没有倾斜？

1、分析节点资源管理器，如果大部分节点已经执行完成，而个别节点长时间执行不完，很可能发生了数据倾斜；

2、分析执行日志，作业在reduce阶段停留在99%，很长时间完成不了，很可能发生了数据倾斜。

三、容易造成数据倾斜的原因

基本业务逻辑是造成数据倾斜的主要原因：

3.1、group by逻辑造成

3.2、distinct count(distinct xx)

3.3、小表join大表

3.4、大表join大表

四、解决方案

4.1、调优参数

set hive.map.aggr=true；

开启map端聚合，效率更高但需要更多的内存

set hive.groupby.skewindata=true;

开启group by数据倾斜时负载均衡，生成的查询计划会有两个MRJob。

第一个MRJob 中，Map的输出结果集合会随机分布到Reduce中，每个Reduce做部分聚合操作，并输出结果，这样处理的结果是相同的GroupBy Key有可能被分发到不同的Reduce中，从而达到负载均衡的目的；

第二个MRJob再根据预处理的数据结果按照GroupBy Key分布到Reduce中（这个过程可以保证相同的GroupBy Key被分布到同一个Reduce中），最后完成最终的聚合操作。

4.2、空值产生的数据倾斜

解决方法1：为空的不参与关联

select * from

--先行过滤，减少读取的数据量

(select * from dmf_dev.dmf_dev__mu_d01 where nvl(t1.id,'') <> '') t1

Inner/left join dmf_dev.dmf_dev__mu_d02 t2

on t1.id = t2.id

union all

select * from dmf_dev.dmf_dev__mu_d01 t1

where nvl(t1.id,'') = ''

解决方法2 ：空值随机赋值方式

select * from dmf_dev.dmf_dev__mu_d01 t1

left join dmf_dev.dmf_dev__mu_d02 t2

on case when nvl(t1.id) = '' then concat('hive',rand()) else t1.id end = t2.id

结论：方法2比方法1效率更好，不但io少了，而且作业数也少了。解决方法1中dev_01读取两次，jobs是2，解决方法2 job数是1 。

4.3、不同数据类型关联产生数据倾斜

场景：两表关联键的数据类型不一致，一个是int，另外一个是string时，默认的Hash操作会按int型的id来进行分配，这样会导致所有string类型id的记录都分配到一个Reducer中。

解决方法：把数字类型转换成字符串类型

select * from dmf_dev.dmf_dev__mu_d01 t1

left join dmf_dev.dmf_dev__mu_d02 t2

on cast(t1.id as string) = t2.id

4.4、小表不小不大，怎么用 map join 解决倾斜问题

set hive.auto.convert.join = true

如果是小表，自动选择map join:

hive.mapjoin.smalltable.filesize=25000000

调大小表的阈值，默认值是25mb

小表过大方案一

参数设置也无法实现mapjoin时，可以先行进行两表内关联结果当做小表，在进行mapjoin；

小表过大方案二

join倾斜， ⼤表join⼩表，但⼩表内存放不下 -> 侧视图（lateral view）⼩表扩容

SELECT

impr.*

,grp.*

FROM (

SELECT

-- 这⾥⽤rand将⼤表的数据打散

CONCAT(group_id,'_',floor(rand()*20)) AS join_key

,*

FROM default.impression_stats_daily_by_hourly

WHERE date='{date}' AND hour='{hour}'

) impr

JOIN (

SELECT

CONCAT(gid,'_',factor) AS join_key

,*

FROM (

SELECT

'0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20' AS number

,gid

,*

FROM default.group_external_properties_hourly

WHERE date_hour='{date}_{hour}'

) extend lateral view explode(split(number, ',')) vt AS factor

-- 这里列转行将小表扩容21倍

) grp

ON

impr.join_key = grp.join_key -- ⽤⼩表扩容的形式保证⼤表数据被打散后还能join上

4.5、大表join的数据倾斜

Map输出key数量极少，导致reduce端退化为单机作业

考虑先对Join中的一个表去重，以此结果过滤无用信息，考虑先对Join中的一个表去重，以此结果过滤无用信息

Map输出key分布不均，少量key对应大量value，导致reduce端单机瓶颈。

总结

让map节点的输出数据更均匀的分布到reduce节点中去，是解决数据倾斜的最终目标。