数据倾斜

1.1 什么是数据倾斜

在map /reduce程序执行时，有时候会出现reduce大部分执行完毕，但是有一个或者几个reduce运行很慢，导致整个程序的处理时间很长，这是因为某一个key的条数比其他key多很多（有时是百倍或者千倍），这条key所在的reduce节点所处理的数据量比其他节点就大很多，从而导致某几个节点迟迟运行不完。这种现象就是因为key分布不均匀、散度不够导致的，也就是数据倾斜。

1.2 数据倾斜产生的原因

在HIVE上执行JOIN,GROUP BY,COUNT, DISTINCT等操作的时候可能会发现ruduce阶段卡在99.99%，一直卡着不能结束，查看任务监控页面，发现只有少量（1个或几个）reduce子任务未完成；进一步查看进程日志或者WEBUI会发现：有一个多几个reduce卡住；CONTAINER报错OOM，读写的数据量极大，超过其它正常的reduce ，伴随着数据倾斜，会出现任务被kill等各种情况。一般情况下Hive的数据倾斜，都发生在SQL中GROUP和ON上，而且和数据逻辑绑定比较深。


数据倾斜的解决办法

2.1 调整配置:

  • 增加JVM 的内存 。这适用于数据的key非常少，往往只能通过硬件的手段来进行调优,增加jvm内存可以显著的提高运行效率。
  • 增加Reduce的个数 。适用于有较多记录值的key都被分到了同一个分区，导致的数据倾斜。如MapReduce，它的分区默认是HashPartitioner，让key的哈希值对设定的Reducer Task个数取余。如果我们增加Reduce的个数（修改numReduceTasks值），就会让一些key被分到不同的分区。虽然工作量仍然会不均衡，但是已不会有这么严重的数据倾斜。
set hive.exec.reducers.bytes.per.reducer=500000000; （500M）
--每个reduce任务处理的数据量，默认为1000^3=1G

set mapred.reduce.tasks = 15;
--每个任务最大的reduce数，默认为999

2.2 聚合类型的数据倾斜解决方法：：

聚合类型出现数据倾斜主要是使用group by、distinct造成的。针对聚合类的数据倾斜，有以下方法：

  • 通过加随机前缀重新设计key 。比如，我们可以在map阶段随机加上一个固定长度的随机数，使得分区的时候不会像之前那样分到同一个节点，完成一次局部聚合。在这之后将前缀去除，重新进行一次全局聚合即可。如下例子：
--水果字段名为category
selectcount (substr(x.category,1,2)) 
from
(select concat(category,'_',cast(round(10*rand())+1as string))
from table1
groupby concat(category,'_',cast(round(10*rand())+1as string))
) x --1阶段聚合
groupby substr(x.category,1,2);     --2阶段聚合
  • 使用COMBINER合并 。Map阶段会将环形缓冲区的数据排序并溢写，在溢写之前，使用combiner将相同key数据进行合并（如累加）。这减轻了数据倾斜的现象，减轻了map端向reduce端发送的数据量(减轻了网络带宽),也减轻了map端和reduce端中间的shuffle阶段的数据拉取数量(本地化磁盘IO速率)。

2.3 大表Join小表的数据倾斜解决办法：

set hive.auto.convert.join = true; -- hive是否自动根据文件量大小，选择将common join转成map join 。
set hive.mapjoin.smalltable.filesize =25000000; --大表小表判断的阈值，如果表的大小小于该值25Mb，则会被判定为小表。则会被加载到内存中运行，将commonjoin转化成mapjoin。一般这个值也就最多几百兆的样子。

当一个大表驱动小表的时候，可以将小表直接读到内存中，进行Map Join，省去了Shuffle阶段。

2.4 大表Join大表的数据倾斜解决办法：

  • 对表分桶排序后Join 。当两个大表都做了分桶处理，且分桶数量相同或者成倍数的时候，可以让相同桶的数据进行Map Join。
set hive.auto.convert.sortmerge.join=true;
set hive.optimize.bucketmapjoin = true;
set hive.optimize.bucketmapjoin.sortedmerge = true;
  • 将大表拆分，对倾斜部分单独处理 。我们可以根据业务中几乎一定会出现数据倾斜的数据拆分出来，将倾斜与未倾斜的部分分别做处理，再将结果合并。

例如，当需要将订单信息表和卖家评级表进行Join，用于获取不同评级的订单量。因为两个表都很大，并且肯定会出现二八法则，即少部分卖家会占有大量买家，而大部分卖家只有很少的买家。在这种情况下，我们可以从订单信息表中将大卖家的评级挑出来放到临时表中，再处理非大卖家的评级，最后做一个union all即可。

部分来源：https://blog.csdn.net/weixin_41812379/article/details/121930949