数据倾斜的原因

1）key分布不均匀，本质上就是业务数据有可能会存在倾斜

2）某些SQL语句本身就有数据倾斜             

关键词	情形	后果
Join	A、其中一个表较小，但是key集中; B、两张表都是大表，key不均	分发到某一个或几个Reduce上的数据远高于平均值
group by	group by 维度过小，某值的数量过多	处理某值的reduce非常耗时
Count(distinct )	某值的数量过多，多次distinct	处理某值的reduce非常耗时，造成数据膨胀

数据产生倾斜的原理：

数据倾斜问题，通常是指参与计算的数据分布不均，即某个key或者某些key的数据量远超其他key，导致在shuffle阶段，大量相同key的数据被发往同一个Reduce，进而导致该Reduce所需的时间远超其他Reduce，成为整个任务的瓶颈。

数据倾斜的表现

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

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

处理数据倾斜

 group by 产生倾斜的问题

前面文章已经提过开启map端聚合，对于开启map端聚合后，数据会现在Map端完成部分聚合工作。这样一来即便原始数据是倾斜的，经过Map端的初步聚合后，发往Reduce的数据也就不再倾斜了。最佳状态下，Map-端聚合能完全屏蔽数据倾斜问题。group by 产生倾斜的问题。

但是并不是所有map端聚合都能完全屏蔽，我们用到结合另外另外一个参数一起使用，开启负载均衡。

开启map聚合：

set hive.map.aggr=true; （默认是 true）

开启map端combiner，减少reduce 拉取的数据量。

set hive.groupby.skewindata=true; （默认是 false）

原理是生成的查询计划会有两个 MR Job。第一个 MR Job 中，Map 的输出结果集合会随机分布到 Reduce 中，每个 Reduce 做部分聚合操作，并输出结果，这样处理的结果是相同的 Group By Key 有可能被分发到不同的 Reduce 中，从而达到负载均衡的目的；第二个 MR Job 再根据预处理的数据结果按照 Group By Key 分布到 Reduce 中（这个过程可以保证相同的 Group By Key 被分布到同一个 Reduce 中），最后完成最终的聚合操作。

实例

explain select province_id                ,count(*) as cnt    from ds_hive.ch12_order_detail_orc t1    group by province_id ;

第一种情况（关闭map聚合和负载均衡），会将3857w数据全部传到reduce阶段进行group by聚合，执行时间60.8s

后台日志说明，map-reduce数据转移处理了8000w条，效率低

第二种（开启map聚合和关闭负载均衡）执行时间26s

后台日志说明，map-reduce数据转移处理了340条，效率高

第三种（开启map聚合和负载均衡）执行时间45s

job1：Map 的输出结果集合会随机分布到 Reduce 中，每个 Reduce 做部分聚合操作，并输出结果

job2:将job1预处理后的数据再进行group by，完成聚合

后台日志说明，job1的map-reduce数据转移处理了340条，但没有完成最终分组聚合，而是将多个reduce数据传给job2的map-reduce进行处理，一定程度上增加了资源的消耗和处理，所以效率比map聚合的一次mr效率低

Join导致的数据倾斜

前文提到过，未经优化的join操作，默认是使用common join算法，也就是通过一个MapReduce Job完成计算。Map端负责读取join操作所需表的数据，并按照关联字段进行分区，通过Shuffle，将其发送到Reduce端，相同key的数据在Reduce端完成最终的Join操作。

如果关联字段的值分布不均，就可能导致大量相同的key进入同一Reduce，从而导致数据倾斜问题。

由join导致的数据倾斜问题，有如下三种解决方案：

map join

使用map join算法，join操作仅在map端就能完成，没有shuffle操作，没有reduce阶段，自然不会产生reduce端的数据倾斜。该方案适用于大表join小表时发生数据倾斜的场景。

相关参数如下：

--启动Map Join自动转换 set hive.auto.convert.join=true; --一个Common Join operator转为Map Join operator的判断条件,若该Common Join相关的表中,存在n-1张表的大小总和<=该值,则生成一个Map Join计划,此时可能存在多种n-1张表的组合均满足该条件,则hive会为每种满足条件的组合均生成一个Map Join计划,同时还会保留原有的Common Join计划作为后备(back up)计划,实际运行时,优先执行Map Join计划，若不能执行成功，则启动Common Join后备计划。 set hive.mapjoin.smalltable.filesize=250000; --开启无条件转Map Join set hive.auto.convert.join.noconditionaltask=true; --无条件转Map Join时的小表之和阈值,若一个Common Join operator相关的表中，存在n-1张表的大小总和<=该值,此时hive便不会再为每种n-1张表的组合均生成Map Join计划,同时也不会保留Common Join作为后备计划。而是只生成一个最优的Map Join计划。 set hive.auto.convert.join.noconditionaltask.size=10000000; hive (default)> create table ds_hive.ch12_order_detail_qingxie_join1 as     SELECT t1.id            ,t2.province_name       from ds_hive.ch12_order_detail_orc t1  left join ds_hive.ch12_province_info_orc  t2         on t1.province_id=t2.id ;

province_id字段是存在倾斜的，若不经过优化，执行日志显示一直卡在99-100直接，单一任务执行了2m35s，这俩种表现都能说明出现了数据倾斜问题，最终执行时间188s

开启mapjoin的自动转换后，可以看出只有map阶段而没有reduce阶段，最终执行时间为26s，相较188s，速度大大增快

skew join

并不是所有的join都能满足mapjoin,这里我们又出现一种优化方案，那就是skew,skew join的原理是，为倾斜的大key单独启动一个map join任务进行计算，其余key进行正常的common join。但是他只能只能某种特定的innerjoin优化，原理图如下：

相关参数如下：

--启用skew join优化 set hive.optimize.skewjoin=true; --触发skew join的阈值，若某个key的行数超过该参数值，则触发 set hive.skewjoin.key=1000000; 这种方案对参与join的源表大小没有要求，但是对两表中倾斜的key的数据量有要求，要求一张表中的倾斜key的数据量比较小（方便走mapjoin）。 set hive.auto.convert.join=true; set hive.optimize.skewjoin=true; explain create table ds_hive.ch12_order_detail_qingxie_join1 as     SELECT t1.id            ,t2.province_name       from ds_hive.ch12_order_detail_orc t1        join ds_hive.ch12_province_info_orc  t2         on t1.province_id=t2.id ;

调整SQL语句

若参与join的两表均为大表，其中一张表的数据是倾斜的，此时也可通过以下方式对SQL语句进行相应的调整。

1）如果是空值关联造成的倾斜，把空值的key变成一随机数(随机值类型需要跟key的类型一致)，把倾斜的数据分到不同的reduce上，由于null值关联不上，处理后并不影响最终结果。注意：join的字段类型一定要一致，否则数据不会分到不同的reduce上。

2）如果热点词，可以考虑分批处理，最后union all在一起。

 count distinct 数据倾斜

我们可以修改对应的sql来进行优化， count+group by 或者sum+group by的方案来优化，在第一阶段选出全部的非重复的字段id，在第二阶段再对这些已消重的id进行计数

其他优化