【使用场景】

　　对RDD执行reduceByKey等聚合类shuffle算子或者在Spark SQL中使用group by语句进行分组聚合时，经过sample或日志、界面定位，发生了数据倾斜。

 

【解决方案】

　　局部聚合+全局聚合，进行两阶段聚合。具体为：

　　将原本相同的key通过附加随机前缀的方式，变成多个不同的key，就可以让原本被一个task处理的数据分散到多个task上去做局部聚合，进而解决单个task处理数据量过多的问题。接着去除掉随机前缀，再次进行全局聚合，就可以得到最终的结果。

• 　　第一步：给key倾斜的dataSkewRDD中每个key都打上一个随机前缀。 

　　例如10以内的随机数，此时原先一样的key，包括集中倾斜的key就变成不一样的了，比如(hello, 1) (hello, 1) (hello, 1) (hello, 1)，就会变成(5_hello, 1) (3_hello, 1) (3_hello, 1) (5_hello, 1) (8_hello, 1) (5_hello, 1) ...  

• 　　第二步：对打上随机前缀的key不再倾斜的randomPrefixRdd进行局部聚合。

　　接着对打上随机数后的数据，执行reduceByKey等聚合操作，进行局部聚合时，就不会数据倾斜了。此时，第一步局部聚合的结果，变成了(5_hello, 3) (3_hello, 2) (8_hello, 1)

• 　　第三步：局部聚合后，去除localAggRdd中每个key的随机前缀。

　　此时，第二步局部聚合的结果，变成了(hello, 3) (hello, 2) (hello, 1)

• 　　第四步：对去除了随机前缀的removeRandomPrefixRdd进行全局聚合。

　　得到最终结果(hello, 6)

 

【方案优点】

　　对于聚合类的shuffle操作导致的数据倾斜，效果不错，通常都可以解决数据倾斜问题，至少大幅缓解数据倾斜，将Spark作业的性能提升数倍以上。

 

【代码实现】

　　代码实现：