Hive 数据倾斜

1.什么是数据倾斜

数据倾斜：数据分布不均匀，造成数据大量的集中到一点，造成数据热点。主要表现为任务进度长时间维持在 99%或者 100%的附近，查看任务监控页面，发现只有少量 reduce 子任务未完成，因为其处理的数据量和其他的 reduce 差异过大。单一 reduce 处理的记录数和平均记录数相差太大，通常达到好几倍之多，最长时间远大于平均时长。

根本原因就是大量相同的key被分配到一个reduce里，造成一个reduce任务累死了，但是其他的reduce任务闲死

2.产生的原因

1，key 分布不均匀。
2，业务数据本身的特性。
3，建表考虑不周全。

4，某些 hsql 语句本身就存在数据倾斜。

就像进行表关联时:

（1）大表join小表：其实小表的key集中，分发到某一个或者几个reduce上的数据远远高于平均值

（2）大表join大表：空值或无意义值：如果缺失的项很多，在做join时这些空值就会非常集中，拖累进度。

（3）group by： group by的时候维度过小，某值的数量过多，处理某值的reduce非常耗时间。

（4）Count distinct：某特殊值过多，处理此特殊值的reduce耗时。

3.具体的数据倾斜以及解决方法:

3.1 空值产生的数据倾斜

使本身为 null 的所有记录替代随机字符串或者先过滤掉空值，而分散到了多个 reduceTask 中了，由于 null 值关联不上，处理后并不影响最终结果。

过滤掉空值的情况

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）赋予空值新的 key 值

select * from log a left outer join user b on
case when a.user_id is null then concat('hive',rand()) else a.user_id end = b.user_id

3.2，不同数据类型关联产生数据倾斜

在join时,两个表关联字段数据类型不一致造成数据向一个MapReduce倾斜

把数字类型 id 转换成 string 类型的 id

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

3.3 join 倾斜

map join 概念：将其中做连接的小表（全量数据）分发到所有 MapTask 端进行 Join，从而避免了 reduceTask，前提要求是内存足以装下该全量数据。自动开启 map join 优化，由两个参数控制：set hive.auto.convert.join=true; //设置 MapJoin 优化自动开启
set hive.mapjoin.smalltable.filesize=25000000 //设置小表不超过多大时开启 mapjoin 优化

1）如何join：关于驱动表的选取，选用join key分布最均匀的表作为驱动表，做好列裁剪和filter操作，以达到两表做join的时候，数据量相对变小的效果。

2）大小表join的时候：使用map join 让小的维度表先进内存，在map端完成reduce。效率很高。

3）大表join大表的时候：把空值的key变成一个字符串加上随机数，把倾斜的数据分到不同的reduce上，由于null值关联不上，处理后不影响最终的结果。

3.4,group by 倾斜

group by的维度太少，某个字段的数量太大，从而导致处理某个值数据的reduce处理非常耗时。

group by 维度过小的时候：采用sum() group by 的方法来替换count(distinct)完成计算。

参数调整：

hive.map.aggr = true    // Map 端部分聚合，相当于Combiner；
hive.groupby.skewindata=true    //有数据倾斜的时候进行负载均衡，
当选项设定为 true，生成的查询计划会有两个 MR Job。
第一个 MR Job 中，Map 的输出结果集合会随机分布到 Reduce 中，
每个 Reduce 做部分聚合操作，并输出结果，
这样处理的结果是相同的 Group By Key 有可能被分发到不同的 Reduce 中，
从而达到负载均衡的目的；
第二个 MR Job 再根据预处理的数据结果按照 Group By Key 分布到 Reduce 
中（这个过程可以保证相同的 Group By Key 被分布到同一个 Reduce 中），
最后完成最终的聚合操作。

3.5,count distinct 倾斜

count distinct 大量相同特殊值，将这些值为空的情况单独处理，如果是计算count distinct，可以不用处理，直接过滤，在最后结果中加1即可。如果还有其他的计算，需要进行group by，可以先将那些值为空的记录单独处理，再和其他计算结果进行 union。

4.数据倾斜解决方法总结

4.1重新分区（Repartitioning）
重新分区（Repartitioning）：重新将表分区，使数据更均匀地分布在各个分区中。可以使用Hive的`INSERT OVERWRITE`语句和`DISTRIBUTE BY`子句来实现。

4.2. 聚合转换（Aggregation Conversion）
聚合转换（Aggregation Conversion）：将一些大的聚合操作拆分成多个小的聚合操作，然后将结果再聚合起来。这样可以减少单个任务要处理的数据量，从而减轻数据倾斜的影响。可以使用Hive的`UNION ALL`语句和子查询来实现。

4.3. 随机分桶（Randomized Bucketing）
随机分桶（Randomized Bucketing）：表分成多个桶（Bucket），并在插入数据时将数据随机分配到这些桶中。这样可以使数据更均匀地分布在各个桶中，从而减轻数据倾斜的影响。可以使用Hive的`CLUSTER BY`子句和`SORT BY`子句来实现。

4.4. 布隆过滤器（Bloom Filter）
布隆过滤器（Bloom Filter）：使用布隆过滤器来过滤掉一些不需要处理的数据，从而减少单个任务要处理的数据量，从而减轻数据倾斜的影响。可以使用Hive的`MAP JOIN`和`STREAMTABLE`功能来实现。

4.5. 动态分区（Dynamic Partitioning）
布隆过滤器（Bloom Filter）：Hive的动态分区功能来动态生成分区。这样可以避免一些特定的分区数据过大而导致的数据倾斜问题。

总之，处理Hive数据倾斜需要根据具体情况采取不同的方法。上述方法仅是一些常见的处理方法，具体选择哪种方法需要根据实际情况进行判断。

其他方法

1，参数调节：hive.map.aggr = true 在map端部分聚合。
2，参数调节：hive.groupby.skewindata=true 数据倾斜时负载均衡。
3，sql语句调节：join时选择key值分布较均匀的表作为驱动表，同时做好列裁剪和分区裁剪，以减少数据量。
4，sql语句调节：大小表join时，小表先进内存。
5，sql语句调节：大表join大表时，把key值为空的key变成一个字符串加上随机数，把倾斜的数据分到不同的reduce上，由于null值关联不上，因此处理后不影响最终结果。