1. 什么是数据倾斜

    简单的讲，数据倾斜就是我们在计算数据的时候，数据的分散度不够，导致大量的数据集中到了一台或者几台机器上计算，造成数据热点问题（数据倾斜的另一种说法），这些数据的计算速度远远低于平均计算速度，导致整个计算过程过慢（木桶效应）。

map阶段数据倾斜现象：
    少数map任务比其他map任务结束的比较晚
reduce阶段数据倾斜现象
    少数reduce任务比其他reduce任务结束的比较晚

2. 造成数据倾斜的原因及解决方案

2.1 map阶段数据倾斜

2.1.1 小文件导致

原因：
    上游表文件的大小特别不均匀，而且小文件特别多，导致当前表Map端读取的数据分布不均匀，引起长尾

解决方案：
    读取的时候合并小文件，尽量做到读取文件大小差不多

设置map输入合并小文件
-- 102400000B=102400KB=100M
-- 每个Map最大输入大小(这个值决定了合并后文件的数量)
set mapred.max.split.size=102400000;
-- 一个节点上split的至少的大小(这个值决定了多个DataNode上的文件是否需要合并)
set mapred.min.split.size.per.node=102400000;
-- 一个交换机下split的至少的大小(这个值决定了多个交换机上的文件是否需要合并)
set mapred.min.split.size.per.rack=102400000;
-- 前3行设置是确定合并文件块的大小，>128M的文件按128M切块，>100M和<128M的文件按100M切块，剩下的<100M的小文件直接合并

-- map执行前合并小文件
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;    

2.1.2 map端聚合导致

原因：
Map端做聚合时，就是combine操作的时候读取文件的某个值特别多而引起长尾

解决方案：
将数据打乱分布

SELECT u.user_name, o.product
FROM users u
JOIN orders o ON u.user_id = o.user_id
DISTRIBUTE BY RAND();

2.2 reduce阶段数据倾斜

2.2.1 key中存在大量的null值或是非正常字符值

原因：
    空值导致的异常

解决方案：
    1.如果是没用的数据，可以过滤掉

select * from log a join user b on a.user_id is not null and a.user_id = b.user_id
union all
select * from log c where c.user_id is null;

    2.如果是有用的数据，可以进行加盐,因为null值参与shuffle时的hash结果是一样的，那么我们可以给null值随机赋值，这样它们的hash结果就不一样，就会进入到不同的Reduce当中去

select * 
from student a
join score b 
on case when a.user_id is null then concat('hive',rand()) else a.user_id=b.user_id;

    3.map端聚合，如果是迭代计算，设置map输出和reduce输出合并小文件

-- 设置map端输出进行合并，默认为true
set hive.merge.mapfiles = true;
-- 设置reduce端输出进行合并，默认为false
set hive.merge.mapredfiles = true;
-- 设置合并文件的大小
set hive.merge.size.per.task = 256*1000*1000;
-- 当输出文件的平均大小小于该值时，启动一个独立的MapReduce任务进行文件merge
set hive.merge.smallfiles.avgsize=16000000;

2.2.2 热点key值过大，导致倾斜

原因：
    热点key值过大，导致倾斜

解决方案：
    1.将热点key加盐打散，加盐局部聚合+去盐全局聚合

select if(user_id like '%1001', substring(user_id, instr(user_id, '-') + 1), user_id) as user_id
     , sum(point_num)                                                                 as point_num
from
    (
        select if(user_id = '1001', concat_ws('-', cast(1 + floor(rand() * 5) as string), user_id), user_id) as user_id
             , sum(point_num)                                                                                as point_num
        from user
        group by if(user_id = '1001', concat_ws('-', cast(1 + floor(rand() * 5) as string), user_id), user_id)
    ) a
group by if(user_id like '%1001', substring(user_id, instr(user_id, '-') + 1), user_id)

    2.将热点key和常规key分开计算

select appid,count(*)
from (
 select md5imei,appid
 from (
    select md5imei,appid,cheat_flags
    from (  select  md5imei,appid,t2.cheat_flags from  imei_open_app  where md5imei="54bc0748b1c0fb46135d117b6d26885e" ) t1
    left outer join cheat_imei  t2
    on t1.md5imei=t2.md5imei
    union all 
    select md5imei,appid,cheat_flags
    from (  select  md5imei,appid,t2.cheat_flags from  imei_open_app  where md5imei!="54bc0748b1c0fb46135d117b6d26885e" ) t11
    left outer join cheat_imei  t12
   on t11.md5imei=t12.md5imei
 )t3
 where t3.cheat_flags is null
) t4
group by  appid  ;

2.2.3 不同数据类型引发的数据倾斜

原因：
    用户表中 user_id 字段为 int，log 表中 user_id 为既有 string 也有 int 的类型， 当按照两个表的 user_id 进行 join 操作的时候，默认的 hash 操作会按照 int 类型的 id 进 行分配，这样就会导致所有的 string 类型的 id 就被分到同一个 reducer 当中

解决：
    把数字类型 id 转换成 string 类型的 id

select * 
from user a 
left outer join log b 
on b.user_id = cast(a.user_id as string)

2.2.4 表关联引发的数据倾斜

原因：
    1.大表join大表导致的数据倾斜

解决：
    1.SMB JOIN
使用条件：
1）两表进行分桶，桶的个数必须相等
2）两边进行join时，join列=排序列=分桶列
比如分桶字段为id，排序字段也为id

使用分桶表，并且开启SMB Join （Sort Merge Bucket Join）
set hive.optimize.bucketmapjoin = true;
set hive.optimize.bucketmapjoin.sortedmerge = true;
set hive.input.format=org.apache.hadoop.hive.ql.io.BucketizedHiveInputFormat;

--创建 bucketed 表
CREATE TABLE users
user_id INT,
user_name STRING
CLUSTERED BY (user_id) INTO 10 BUCKETS;

CREATE TABLE orders
order id INT,
user id INT,
order_amount DOUBLE
CLUSTERED BY (user_id) INTO 10 BUCKETS;

-- 插入示例数据
INSERT INTO users VALUES (1, 'Alice'), (2, 'Bolb'),(3,'Charlie');
INSERT INTO orders VALUES (101,1, 250.50),(1002,2,150.75),(103,3,200.00);

--执行 SMB Join
SET hive.optimize.bucketmapjoin = true;
SET hive.exec.dynamic.partition.mode = nonstrict;
SELECT u.user_name,o.order_amount
FROM users u
JOIN orders o ON u.user_id = o.user_id;                    

原因：
    2.大表join小表导致的数据倾斜

解决：
    1.mapjoin 将其中做连接的小表（全量数据）分发到所有 MapTask 端进行 Join，从而避免了 ReduceTask，前提要求是内存足以装下该全量数据。

-- 开启自动mapjoin
set hive.auto.convert.join = true;
-- 是否使用原生的向量化执行模式执行MapJoin，它会比普通MapJoin速度快。默认值为False。
set hive.vectorized.execution.mapjoin.native.enabled = true;
-- 大表小表的阈值设置（默认25M一下认为是小表）
set hive.mapjoin.smalltable.filesize=26214400;

解决：
    2.在执行MR的时候, 会动态统计每一个 k2的值出现重复的次数, 当这个重复的次数达到一定的阈值后, 认为当前这个k2的数据存在数据倾斜, 自动将其剔除, 交由给一个单独的MR来处理即可,两个MR处理完成后, 将结果基于union all 合并在一起即可

 -- 开启运行期处理倾斜参数默认false
set hive.optimize.skewjoin=true; 
 -- 阈值,  此参数在实际生产环境中, 需要调整在一个合理的值(否则极易导致大量的key都是倾斜的),默认100000
set hive.skewjoin.key=100000;  

解决：
    3. 在创建这个表的时候, 我们就可以预知到后续插入到这个表中数据, 那些key的值会产生倾斜, 在建表的时候, 将其提前配置设置好即可, 在后续运行的时候, 程序会自动将设置的key的数据单独找一个MR来进行处理即可, 处理完成后, 再和原有结果进行union all 合并操作

set hive.optimize.skewjoin.compiletime=true;  -- 开启编译期处理倾斜参数
 
 CREATE TABLE list_bucket_single (key STRING, value STRING)
 -- 倾斜的字段和需要拆分的key值
 SKEWED BY (key) ON (1,5,6)
 -- 为倾斜值创建子目录单独存放
 [STORED AS DIRECTORIES];

解决：
    4. 大量的key可以拼接rand() 适用于: 提前知道那些key存在倾斜

如果大量相同的男,可以拼接随机数: concat('男',rand())

2.2.5 count(distinct xx)导致数据倾斜

原因：
    数据key值大的键分到一个reduce,任务长尾

解决：
    1.开启count distinct的数据倾斜参数优化

set hive.optimize.countdistinct = true;

解决：
    2.先group by 再进行计数

select count(1) as dis_cnt 
from 
(select age 
 from user_info 
 group by age
 ) t

2.2.6 group by出现数据倾斜

原因：
    数据key值大的键分到一个reduce,任务长尾

解决：
    1.开启group by的数据倾斜参数优化
    当有数据倾斜的时候进行负载均衡，生成的查询计划有两个MapReduce任务，第一个MR Job中，Map的输出结果会随机分布到Reduce中，每个Reduce做部分聚合操作，并输出结果，这样处理的结果是相同的Group By Key有可能被分发到不同的Reduce中，从而达到负载均衡的目的；第二个MR Job再根据预处理的数据结果按照Group By Key分布到Reduce中（这个过程可以保证相同的Group By Key被分布到同一个Reduce中），最后完成最终的聚合操作

set hive.groupby.skewindata = ture;

    2.将key值随机打散

select key_name    
	,sum(sum_val) as sum_val
from 
(	-- 第一个mr，添加随机数列打散
	select key_name
		,ceiling(rand() * 99) as rnk    -- 添加随机数打散
		,sum(val) as sum_val
	from table_name 
	group by key_name,ceiling(rand() * 99)
) t 
group by key_name   -- 再次聚合，第二个mr