[大数据]大数据中数据倾斜问题

首先，需要了解一下什么是数据倾斜。简单来说就是在job任务执行的过程中，某些分区或者节点上的数据，明显高于其他分区或节点的数据，导致这部分的数据处理任务比其他任务要大很多，从而成为这个阶段执行最慢的部分，进而成为整个作业执行的瓶颈，甚至直接导致作业失败，即数据分配严重不均匀，导致job任务失败。

举个简单的例子就是，一个job当中有1000个task，其中998个task1分钟完成了，剩下2个task10分钟还没有完成，最终导致了job失败。

数据倾斜最终会导致job失败，而在那之前，其实还有其他的危害。

1、任务会长时间的挂起，资源利用率下降

因为不管是mr任务还是spark任务，计算作业都是分阶段进行的，阶段和阶段之间通常都会有数据上的通信交互，比如mr任务，reduce任务的数据明显需要等待map任务的执行完毕才会开始，spark也有阶段划分。也就是说，当产生数据倾斜导致任务执行时长过长，或者直接挂起，那么后续的阶段将一直处于等待状态，整个作业也就一直挂起。这个时候，资源被这个作业占据，但是却只有极少数task在执行，造成计算资源的严重浪费，利用率下降。

2、引发内存溢出，导致任务失败

数据发生倾斜时，可能导致大量数据集中在少数几个节点上，在计算执行中由于要处理的数据超出了单个节点的能力范围，最终导致内存被撑爆，报OOM异常，直接导致任务失败。

3.作业执行时间超出预期，导致后续依赖数据结果的作业出错

有时候作业与作业之间，并没有构建强依赖关系，而是通过执行时间的前后时间差来调度，当前置作业未在预期时间范围内完成执行，那么当后续作业启动时便无法读取到其所需要的最新数据，从而导致连续出错。

那么，产生数据倾斜的原因又是什么呢？

1、数据读取的时候就是倾斜的

最简单的原因，如果数据一开始就是倾斜的，那么不管怎么样，计算的时候肯定会产生数据倾斜。比如说，在读入阶段的时候，可能partition分配不均匀的数据。

2.shuffle产生倾斜

在shuffle阶段造成倾斜，在实际的工作中更加常见，比如特定key值数量过多，导致join发生时，大量数据涌向一个节点，导致数据严重倾斜，个别节点的读写压力是其他节点的好几倍，容易引发OOM错误。

3.过滤导致倾斜

在某些情况下，有可能原本分区数据是均匀的，但是由于经过了过滤操作，导致过滤了大量的数据，从而导致了数据的倾斜。

虽然可能过滤掉的数据一般可能不会造成OOM，但是倾斜的数据可能会随着计算的逐渐累积，最终引发任务失败。

最后就是解决数据倾斜的办法啦！

1、尽量保证数据源是均衡的

虽然这个是一句废话，但是在不同阶段产生的数据的时候，都尽量需要保证上游的作业不要给下游任务埋坑……在程序输出写入文件的时候，尽量不要使用spark中的coalesce，而是使用repartition，因为coalesce虽然可以减少分区，但是是在尽量保证分区不变的情况下，进行缩减分区，coalesce不会进行shuffle，而repartition会进行shuffle，从而使得数据均匀。

经过过滤操作的大数据集最好也是在结束后，进行一次repartition，因为在数据集进行过滤之后，可能存在数据倾斜的问题。所以，进行一次repartition让数据集重回均匀分布的状态。

2、对小表进行广播

如果大表和小表进行join的操作，尽量需要避免reduce端的join，可以将小表进行广播到每个节点，用时间换空间，实现map端join，从而省去了shuffle。

3、join前优化

如计算是先join后根据key聚合，那可以改为先根据key聚合然后再join。又如，需求是join后做distinct操作，在不影响结果的前提下，可以改为先distinct，然后再join。这些措施都是可以有效避免重复key过多导致join时倾斜。

4、Group By

在默认的情况下，Map阶段同一个key的数据会分发给一个reduce，当一个key数据过大时就会产生数据倾斜。可以开启Map端端进行聚合，默认为 True。set hive.map.aggr = true。有数据倾斜的时候进行负载均衡（默认是 false），set 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、Count(Distinct) 去重统计

数据量小的时候无所谓，数据量大的情况下，由于 COUNT DISTINCT 操作需要用一个 Reduce Task 来完成，这一个 Reduce 需要处理的数据量太大，就会导致整个 Job 很难完成， 一般 COUNT DISTINCT 使用先 GROUP BY 再 COUNT 的方式替换,但是需要注意 group by 造成 的数据倾斜问题。虽然会多用一个 Job 来完成，但在数据量大的情况下，这个绝对是值得的。

6、小文件进行合并

可以在输入的时候使用CombineInputFormat对小文件进行聚合，减少map数量，在hive当中可以设置set hive.input.format= org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;或者使用har进行归档存入hdfs当中。