16、Hive数据倾斜与解决方案

数据倾斜

1、什么是数据倾斜

由于数据分布不均匀，造成数据大量的集中到一点，造成数据热点

2、数据倾斜的现象

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

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

3、数据倾斜的情况

创新互联公司网站建设公司一直秉承“诚信做人，踏实做事”的原则，不欺瞒客户，是我们最起码的底线！ 以服务为基础，以质量求生存，以技术求发展，成交一个客户多一个朋友！专注中小微企业官网定制，成都网站建设、网站建设，塑造企业网络形象打造互联网企业效应。

4、数据倾斜的原因

1)、key分布不均匀

2)、业务数据本身的特性

3)、建表时考虑不周

4)、某些SQL语句本身就有数据倾斜

5、数据倾斜的解决方案

5.1 map端聚合

--Map 端部分聚合，相当于Combiner
hive.map.aggr = true；
--有数据倾斜的时候进行负载均衡
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 中），最后完成最终的聚合操作。

5.2 SQL语句调节

• 如何Join

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

• 大小表Join

  使用map join让小的维度表（1000条以下的记录条数） 先进内存。在map端完成reduce.

• 大表Join大表

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

• count distinct大量相同特殊值

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

• group by维度过小

  采用sum() group by的方式来替换count(distinct)完成计算。

• 特殊情况特殊处理

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

5.3 典型的业务场景

• 空值产生的数据倾斜

  • 场景

  如日志中，常会有信息丢失的问题，比如日志中的 user_id，如果取其中的 user_id 和 用户表中的user_id 关联，会碰到数据倾斜的问题。

  • 解决办法

  --user_id为空的不参与关联
  
  select * from log a
  join users b
  on a.user_id is not null
  and a.user_id = b.user_id
  union all
  select * from log a
  where a.user_id is null;
  
  --赋与空值分新的key值
  select *
  from log a
  left outer join users b
  on case when a.user_id is null then concat(‘hive’,rand()) else a.user_id end = b.user_id;
  

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

  • 场景

  用户表中user_id字段为int，log表中user_id字段既有string类型也有int类型。当按照user_id进行两个表的Join操作时，默认的Hash操作会按int型的id来进行分配，这样会导致所有string类型id的记录都分配到一个Reducer中。

  • 解决办法

  • 把数字类型转换成字符串类型

  select * from users a
    left outer join logs b
    on a.usr_id = cast(b.user_id as string);