# 数据倾斜
## 4.1. 什么是数据倾斜，现象是什么？
所谓数据倾斜(data skew)，其实说白了，由于数据分布不均匀造成计算时间差异很大，产生了一些列异常现象。
常见的现象有两种：
1. 个别task作业运行缓慢
大多数的task运行都很快速，但是极个别的task运行非常缓慢，甚至是正常task运行时间好多倍。
莫名其妙的OOM异常
这是一种相对比较少见的现象，正常运行的task作业，突发发生了一个OOM异常。但是这只是一种假设，需要验证，因为流量的突然暴增也会经常导致OOM异常。
发生数据倾斜的原因是什么
宽依赖了，也就是shuffle操作。shuffle操作说白了就是将key相同的数据经过网络拉取到同一个节点上的同一个partition中进行聚合操作。加入绝大多数key都对应10条数据，但是有个别key对应的数据量10w条，所以经过shuffle操作之后，这特殊key的10w+的数据会到一个分区中去，而其它分区数据量相对正常，可不就造成了个别task任务执行时间是其它普通task的若干倍。
请问，shuffle操作是发生数据倾斜的什么条件？
必要条件。
## 4.3. 如何解决数据倾斜
### 4.3.1. 解决的思路
我们已经知晓发生的原因是由于某些key对应的数据量过多操作的，所以我们首先需要找到这些key，问题在于如何找到这些key？显然不能基于全量的数据找，只能抽样，使用sample算子进行处理。
找到这些key之后，需要进行分拆（最常见的处理思路：加随机前缀），最后进行全局处理。
### 4.3.2. 在Hive ETL中做预处理
这个处理的方法，主要在于Spark作业加载hive表中的数据，进行业务处理。假如hive的数据有倾斜现象，在spark中的处理，自然会出现dataskew。而如果spark作业一般只是想web 端提供查询服务，针对这种情况就比较适合这个解决方法。
Hive ETL预处理，数据倾斜的现象在hive中提前被处理，这样加载到spark中的数据有倾斜吗？没有！此时spark给web服务端只提供一个查询服务，所以没有的数据倾斜，效率非常高！只不过此时将数据倾斜解决掉了吗？是把spark端的dataskew转移到hive中。
### 4.3.3. 过滤掉发生数据倾斜的key
找到哪些发生数据倾斜的key，同时**必须要想业务人员确认这些key是否有用**，如果没用直接使用filter算子过滤掉就行。
以后在工作中，切忌，但凡是删除、过滤、更新等待操作，一定慎重。
### 4.3.4. 提高程序并行度
程序运行缓慢，第一反应大多是资源分配不足，并行度不够。提高并行度是我们做数据倾斜调优的第一步尝试，**提高并行度会在一定程度上减轻数据倾斜的压力，但是并不能从彻底上根除数据倾斜**。因为一旦发生数据倾斜，倾斜的key无论如何提高并行度，经过shuffle操作都会直到一个分区中去。
如何提高并行度？两个地方进行设置。
1. spark.default.parallelism 设置spark程序全局并行度
2. shuffle操作的第二个参数进行设置(局部)并行度
/%E6%96%87%E6%A1%A3/assets/1561346170884.png)
/%E6%96%87%E6%A1%A3/assets/1561346192356.png)
### 4.3.5 进行两阶段聚合
#### 4.3.5.1. 原理示意图
两阶段聚合操作，指的是局部聚合+全局聚合。该方法适合于哪些XxxxByKey的操作，比如groupByKey、reduceByKey的聚合操作。
#### 4.3.5.2. 代码实现
```scala
object _01TwoStageDataskewOps {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
.setAppName("_01TwoStageDataskewOps")
.setMaster("local[2]")
val sc = new SparkContext(conf)
val list = List(
"hello hello hello hello you you hello",
"hello hello hello you you hei hei hello hello hello"
)
val listRDD = sc.parallelize(list)
val pairsRDD:RDD[(String, Int)] = listRDD.flatMap(line => {
line.split("\\s+")
}).map((_, 1))
//step 1、确认发生数据倾斜的key --->sample算子
val sorted = pairsRDD.sample(true, 0.6).countByKey().toList.sortWith((m1, m2) => m1._2 > m2._2)
println("抽样数据排序之后的结果：")
println(sorted.mkString("\n"))
//step 1.1 获取发生数据倾斜key
val dataskewKey = sorted.head._1
println("发生数据倾斜的key为：" + dataskewKey)
//step 2、加随机前缀
val prefixPairsRDD = pairsRDD.map{case (word, count) => {
if(word == dataskewKey) {
val random = new Random()
val prefix = random.nextInt(2)
(s"${prefix}_${word}", count)
} else {
(word, count)
}
}}
println("=============step 2、加随机前缀数据===================")
prefixPairsRDD.foreach(println)
//step 3、局部聚合
val partAggrRDD:RDD[(String, Int)] = prefixPairsRDD.reduceByKey(_+_)
println("=============3、局部聚合结果===================")
partAggrRDD.foreach(println)
//step 4、去掉随机前缀
val unPrefixPairsRDD = partAggrRDD.map{case (word, count) => {
if(word.contains("_")) {
(word.substring(2), count)
} else {
(word, count)
}
}}
println("=============step 4、去掉随机前缀的数据===================")
unPrefixPairsRDD.foreach(println)
//step 5、进行全局聚合
val fullAggrRDD = unPrefixPairsRDD.reduceByKey(_+_)
println("=============step 5、进行全局聚合===================")
fullAggrRDD.foreach(println)
sc.stop()
}
}
```
### 4.3.6. 使用map-join代替reduce-join
这个操作主要是针对join类的聚合操作，多表关联，前提条件是大小表关联。
所谓reduce-join操作就是很直白的调用join算子，执行操作，这个过程是有shuffle的。
所谓map-join操作呢，将小表广播到各个executor，在map类算子中完成关联操作。
这个操作，请问，从根本上解决了数据倾斜了没有？从根本上解决了数据倾斜，因为有map-join代替reduce-join没有shuffle操作，肯定就没有数据倾斜了。
参见之前广播变量使用的代码。
### 4.3.7. 使用采样key并分拆进行聚合
join操作的是两张大表，一张表正常，一张表中有个别key异常，其余正常。怎么办？
#### 4.3.7.1. 原理示意图
#### 4.3.7.2. 代码实现
```scala
object _02SplitKeyExtendOps {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
.setAppName("_01TwoStageDataskewOps")
.setMaster("local[2]")
val sc = new SparkContext(conf)
val left = List(
("hello", 1),
("hello", 2),
("hello", 3),
("you", 1),
("me", 1),
("you", 2),
("hello", 4),
("hello", 5)
)
val right = List(
("hello", 11),
("hello", 12),
("you", 11),
("me", 12)
)
val leftListRDD:RDD[(String, Int)] = sc.parallelize(left)
val rightListRDD:RDD[(String, Int)] = sc.parallelize(right)
//step 1、采样找到异常的key
val sampledRDD = leftListRDD.sample(true, 0.8)
val sorted = sampledRDD.countByKey().toList.sortWith((m1, m2) => m1._2 > m2._2)
println("排序之后的采样数据：" + sorted.mkString("\n"))
val dataskewKey = sorted.head._1
/*
step 2、根据异常的key将左右表都拆分正常的数据和异常的数据
*/
val dsLeftRDD:RDD[(String, Int)] = leftListRDD.filter{case (word, count) => word == dataskewKey}
val commonLeftRDD:RDD[(String, Int)] = leftListRDD.filter{case (word, count) => word != dataskewKey}
val dsRightRDD:RDD[(String, Int)] = rightListRDD.filter{case (word, count) => word == dataskewKey}
val commonRightRDD:RDD[(String, Int)] = rightListRDD.filter{case (word, count) => word != dataskewKey}
println("step 2、根据异常的key将左右表都拆分正常的数据和异常的数据")
println("左表异常数据：")
dsLeftRDD.foreach(println)
println("左表正常数据：")
commonLeftRDD.foreach(println)
//step 3、对左表异常数据添加N以内的随机前缀
println("step 3、对左表异常数据添加N以内的随机前缀")
val prefixLeftRDD = dsLeftRDD.map{case (word, count) => {
val random = new Random()
val prefix = random.nextInt(2)
(s"${prefix}_${word}", count)
}}
prefixLeftRDD.foreach(println)
//step 4、对右表异常数据进行N倍的扩容
println("step 4、对右表异常数据进行N倍的扩容")
val prefixRightRDD = dsRightRDD.flatMap{case (word, count) => {
val ab = ArrayBuffer[(String, Int)]()
for(i <- 0 until 2) {
ab.append((s"${i}_${word}", count))
}
ab
}}
prefixRightRDD.foreach(println)
/**
* step 5、分别对异常数据和正常数据进行join操作
*/
println("step 5、分别对异常数据和正常数据进行join操作")
val commonJoinedRDD = commonLeftRDD.join(commonRightRDD)
val dsJoinedRDD = prefixLeftRDD.join(prefixRightRDD)
println("5.1 正常数据join的结果")
commonJoinedRDD.foreach(println)
println("5.2 异常数据join的结果")
dsJoinedRDD.foreach(println)
//5.3 去掉异常数据的随机前缀
val dsFinalJoinedRDD = dsJoinedRDD.map{case (word, count) => {
(word.substring(word.indexOf("_") + 1), count)
}}
//step 6、全局的union操作，前提是去掉step5中的异常数据的前缀
val finalJoinedRDD = dsFinalJoinedRDD.union(commonJoinedRDD)
println("step 6、全局的union操作，前提是去掉step5中的异常数据的前缀")
finalJoinedRDD.foreach(println)
sc.stop()
}
}
```
扩展，两张大表，左表全量异常，右表正常。
没有好的解决方案，左表全量加N以内的随机前缀，右表全量进行N倍的扩容。可能会有的问题，扩容之后的存储压力非常大，可能发生OOM异常。
这个方案4.3.7，实际上是以空间换时间。
### 4.3.8. 多种手动多管齐下
如果说上述的单一操作解决不了问题怎么办？那就一起上！
