spark关于数据倾斜问题

spark的数据倾斜调优方案归纳总结：

不来虚的，直接上解决方法。

数据倾斜产生原因：在运算过程中把数据分配给不同的Task，一般需要Shuffle过程，同一个Key都会交给Task处理，但是有时同一个Key的values数据量太多造成数据堆积等。

判断是否发生数据倾斜：通过Web UI查看Task的metrics某些Task相当于其他大多数Task来说所消耗相当长的时间。

数据倾斜解决方案：

（1）对源数据进行聚合并过滤掉导致倾斜的Keys。

对源数据进行聚合并过滤掉导致倾斜的keys：数据源记录中导致数据倾斜的Key可能只有几个，这些Key不影响业务的计算，汇聚后对应的Value值却很多，可能导致数据倾斜。

1. 1. 可使用filter算子过滤掉倾斜的Key值；
   2. 若使用Hive工具，可以将数据倾斜的操作前移到Hive中进行，在Hive中对数据倾斜进行预处理，从数据根源上解决数据倾斜；
   3. 若在Spark SQL中发生倾斜，可使用where/filter算子过滤，然后再进行groupBy等操作。或采用sample算子，对DataFrame的Key值采样，若不是业务需要，使用where/filter直接过滤掉。

（2）适当提高Reducer端的并行度（并行度指的是RDD的分区数）。

改善并行度之所以改变数据倾斜的原因：若某个Task有100个Key且数据量特别大，就有可能导致OOM（Out Of Memory）或任务运行特别慢，若把并行度调大，则可分解该Task的数据量。

2.1对于reduceByKey可传入并行度参数numPartitions: Int，该参数设置了shuffle算子执行Shuffle read Task的数量；

2.2 亦可自定义Partitioner，增加Executor改变计算资源。

（3）使用随机Key实现双重聚合

随机Key双重集合：指Spark分布式计算对RDD调用reduceByKey各算子，使用对Key值随机数前缀的处理技巧，对Key值进行二次聚合。第一次聚合（局部聚合）是对每个Key值加上一个随机数，执行第一次reduceByKey聚合操作；第二次聚合（双重聚合）是指去掉Key值的前缀随机数，执行第二次reduceByKey聚合，得到最终结果。

原理：使用随机Key实现双重聚合，原理剖析算子reduceByKey。假设有数据倾斜，则给所有的Key加上一个随机数，然后进行reduceByKey操作，原来非常大的倾斜的Key就分而治之变成若干个较小的Key，不过此时结果和原来不一样，则使用map操作把随机数前缀去掉，然后再次进行reduceByKey操作。

注意事项：使用随机Key双重聚合局限于单个RDD的reduceByKey、groupByKey等算子。若两个RDD的数据量都特别大，而且倾斜的Key特别多，就无法使用随机Key双重聚合方案。

（4）使用Mapper端进行join操作

把Reducer端的操作放在Mapper端执行，就避免了Shuffle，很大程度上化解掉了数据倾斜问题。RDD是链式操作，DAGScheduler根绝RDD的不同类型的依赖关系（宽依赖和窄依赖）划分成不同的Stage，当发生宽依赖时把Stage划分成更小的Stage，宽依赖算子如reduceByKey、groupByKey等，在Mapper端进行join操作就是把宽依赖减掉，避免shuffle操作。

场景应用：若两个RDD进行操作，其中一个RDD数据量较小，就把此RDD的数据以广播变量的形式散发给各个Executor，这样就可与另一个RDD进行map操作了。

条件：要进行join的RDD，其中一个RDD数据量比较小。

（5）对倾斜Keys采样后进行单独的join操作

若果两个RDD的数据量都比较大，则考虑使用采样方案。

在spark中可直接使用采样算法sample找出哪个RDD导致数据倾斜，sample算法采样一般不超过原数据的30%

对倾斜的Keys进行采样步骤：

RDD1和RDD2进行join操作，采用采样的方式返现RDD1有严重的数据倾斜的Key。

第一步：采用RDD提供的采样接口，对RDD1全部原数据进行采样，然后基于采样的数据可以计算出哪些Key的Values值最多；

第二步：把全部数据分为两部分RDD11和RDD12，RDD11代表导致倾斜的Key，RDD12是不会发生数据倾斜的Key；

第三步：RDD11和RDD12分别于RDD2进行join操作，然后把join的结果进行union操作，从而得出RDD1与RDD2直接join的结果。

原理：RDD12与RDD2进行join操作不会发生数据倾斜，因为两个RDD中的Key的Values值都不多，RDD11和RDD2进行join操作，利用spark core的并行机制对RDD11的Key的数据进行拆分。

注意事项：对RDD1进行sample采样计算出一个结果，采样后进行map操作，通过reduceByKey操作计数，然后对Key和Value进行置换，通过SortByKey进行排序，再进行map置换操作，从而找出哪个Key值倾斜严重，对其进行过滤，提取到RDD11中，没有发生倾斜的提取到RDD12中。

（6）使用随机数进行join

倾斜的Key加上随机数：对Key进行map操作加上随机数，计算结果后再次进行map操作，将随机数去掉，这样得到的结果与不加随机数的结果是一样的，好处是可控制并行度。

使用随机数的步骤：

第一步：对RDD1使用mapToPair算子进行转换，使用Random random = new Random();构建一个随机数，如任意一个100以内的随机数random.nextInt(100)赋值给prefix，将prefix与Key值拼接成一个新的Key值：prefix_Key；

第二步：对RDD2使用mapToPair算子进行转换，对于RDD1相同的Key值，同样构建随机数加上前缀，也将prefix加上Key值拼接成新的key值：prefix_Key；

第三步：RDD1和RDD2根绝prefix_Key进行join操作；

第四步：RDD1和RDD2进行join的结果再次map算子转换，去掉随机数，得到最终结果。

使用场景：两个RDD的某一个Key或多个key值的数据量特别大，在进行join等操作时发生数据倾斜，必须知道哪些Keys的值发生倾斜。

（7）通过扩容进行join