HIVE 数据倾斜浅谈_hive join数据倾斜

HIVE 数据倾斜浅谈

一、数据倾斜现象

  map100%,reduce一直卡在一个值，如99%。

二、数据倾斜的原因

  数据按key的hash值分配到reduce中，如果有的key值比较集中，就会导致某个或某些reduce分配的数据量太大，这样当其他reduce运行完毕，分配数据量过大的reduce仍在运行，导致reduce进度一直卡着不动，这种现象被称为数据倾斜。造成数据倾斜的根本原因就是key值分布不均匀。

三、数据倾斜的操作

join、group by、Count Distinct

四、数据倾斜的处理方法

1. 参数配置方案

1.groupby、count distinct数据倾斜：

hive.map.aggr = true
hive.groupby.skewindata=true
1
2

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

hive.optimize.skewjoin=true；
1

  如果join过程中出现了数据倾斜，应该设置为true

set hive.skewjoin.key=100000；
1

  这个是join的键对应的记录条数超过这个值则会进行优化，优化措施：正常是只有一个job的，优化后会有两个job。当倾斜Key值达到100000以上的时候，hive会把不倾斜的key进行join，倾斜的 key 的数据将会被写入HDFS临时文件，hive会再启动一个job，然后将倾斜的数据进行map 端join，最终将两个结果合并。
  但是这种处理方式不是所有地方都有效。

2. sql优化方案

1. count distinct、group by

• 当group by key为空时，如果只是计算count
  distinct，可以不用处理，直接过滤，在最后结果中加1。如果还有其他计算，需要进行group
  by，可以先将值为空的记录单独处理，再和其他计算结果进行union。
• count(distinct user_id)，采用

select count(1) from (select user_id from log group by user_id)  t;
1

来替换count(distinct)完成计算。

• 在业务逻辑优化效果的不大情况下，有些时候是可以将倾斜的数据单独拿出来处理。最后union回去。

2. 空值产生的数据倾斜

1.把空值单独处理，再与其他值的处理结果进行uinon all.
2.把空值的key变为随机数。

select *
  from log a
  left join users b
  on case when a.user_id is null then concat(‘hive’,rand() ) else a.user_id end = b.user_id;
1
2
3
4

  方法2比方法1效率更好，不但io少了，而且作业数也少了。解决方法1中 log读取两次，jobs是2。解决方法2 job数是1 。这个优化适合无效 id (比如 -99 , ’’, null 等) 产生的倾斜问题。把空值的 key 变成一个字符串加上随机数，就能把倾斜的数据分到不同的reduce上 ,解决数据倾斜问题。

3. 条件中等号两端数据类型不同

  类型转换过程中可能会出现转换失败，比较常见的是string转int，失败后会造成大量null值，进而流入同一个reduce，导致数据倾斜。
  处理办法就是在条件中进行类型转换，保证两端数据类型一致。

4.小表关联大表

  Map端join会使所有的 maptask 节点都装载小表 b 的所有数据，然后大表 a 的 一个数据块数据比如说是 a1 去跟 b 全量数据做链接。实现方法：

select /* +mapjoin(a) */ a.id aid, name, age from a join b on a.id = b.id;
1

  在 hive0.11 版本以后会自动开启 map join 优化，由两个参数控制：

set hive.auto.convert.join=true; //设置 MapJoin 优化自动开启
set hive.mapjoin.smalltable.filesize=25000000 //设置小表不超过多大时开启 mapjoin 优化
1
2

5.小表不小关联大表——处理大表文件，使其变为可加载到内存的小表

  可以根据业务逻辑，只取必要数据(行、列两个维度限制)，尽量减小数据大小，使之成为小表。比如：
当分析某段时间的日志表logs数据是亿级的，用户表users数目有数十万，把users当小表做map join会超出内存时，我们可以把users行数减少到只取出logs中包含的users，然后进行列裁剪，这样可能大大缩减users所占空间，从而可以实现map join。

select /*+mapjoin(t1)*/* from log t1
left outer join (
 select /*+mapjoin(t11)*/ d.*
 from ( select distinct user_id from log ) t11 join users t12 on t11.user_id = t12.user_id
) t1
on t0.user_id = t1.user_id;
1
2
3
4
5
6

6.大表关联大表

  两个Hive表进行join的时候，如果数据量都比较大，那么此时可以看一下两个Hive表中的key分布情况。

6.1 如果出现数据倾斜，是因为其中某一个Hive表中的少数几个key的数据量过大，而另一个Hive表中的所有key都分布比较均匀

方案实现思路：

• 对包含少数几个数据量过大的key的那个表，通过sample算子采样出一份样本来，然后统计一下每个key的数量，计算出来数据量最大的是哪几个key。
• 然后将这几个key对应的数据从原来的表中拆分出来，形成一个单独的表，并给每个key都打上n以内的随机数作为前缀；
• 接着将另一个表，也过滤出来这几个key对应的数据并形成一个单独的表，将每条数据膨胀成n条数据，这n条数据都按顺序附加一个0~n的前缀。
• 再将附加了随机前缀的独立表与另一个膨胀n倍的独立表进行join，此时就可以将原先相同的key打散成n份，分散到多个task中去进行join了。
• 不倾斜的数据照常join即可。
• 最后将两个join的结果使用union all算子合并起来即是最终结果。
• 比较集中的key值如果不想做数据膨胀，也可以使用map join将记录较少的表加载到内存中。

方案优点：对于join导致的数据倾斜，如果只是某几个key导致了倾斜，采用该方式可以用最有效的方式打散key进行join。而且只需要针对少数倾斜key对应的数据进行扩容n倍，不需要对全量数据进行扩容。避免了占用过多内存。
方案缺点：如果导致倾斜的key特别多的话，比如成千上万个key都导致数据倾斜，那么这种方式也不适合。

方案二：分桶join+map join
参见：https://blog.csdn.net/Samaritan_H/article/details/79090103

6.2 如果出现数据倾斜的key比较多

方案一：我们可以按6.1的方式对倾斜表所有数据加上n以随机前缀，对另一张表进行全表数据膨胀操作，然后进行关联。
这样可以缓解数据倾斜问题。但是比较消耗内存资源。
方案二：依旧分离出比较集中的key值记录，另一张表也取出相应的key值记录，然后开启map join进行关联操作，对结果进行uion。
方案三：分桶join+map join
参考文献：https://blog.csdn.net/jin6872115/article/details/79878391