0. 什么是数据倾斜
对于集群系统，一般缓存是分布式的，即不同节点负责一定范围的缓存数据。我们把缓存数据分散度不够，导致大量的缓存数据集中到了一台或者几台服务节点上，称为数据倾斜。一般来说数据倾斜是由于负载均衡实施的效果不好引起的。
来源百度百科
对于数据计算过程来说，数据倾斜指的是，并行处理的数据集中，某一部分（如Spark或Kafka的一个Partition）的数据显著多于其它部分，从而使得该部分的处理速度成为整个数据集处理的瓶颈。
1. 数据倾斜的现象
多数task执行速度较快,少数task执行时间非常长，或者等待很长时间后提示你内存不足，执行失败。
2. 数据倾斜的影响
1）数过多的数据在同一个task中执行，将会把executor撑爆，造成OOM，程序终止运行。,据倾斜直接会导致一种情况：Out Of Memory。
2）运行速度慢 ,spark中一个stage的执行时间受限于最后那个执行完的task，因此运行缓慢的任务会拖累整个程序的运行速度（分布式程序运行的速度是由最慢的那个task决定的）。要是发生在Shuffle阶段。同样Key的数据条数太多了。导致了某个key(下图中的80亿条)所在的Task数据量太大了。远远超过其他Task所处理的数据量。
一个经验结论是：一般情况下，OOM的原因都是数据倾斜\
3. 如何定位数据倾斜
数据倾斜一般会发生在shuffle过程中。很大程度上是你使用了可能会触发shuffle操作的算子：distinct、groupByKey、reduceByKey、aggregateByKey、join、cogroup、repartition等。
原因： 查看任务-》查看Stage-》查看代码
某个task执行特别慢的情况
某个task莫名其妙内存溢出的情况
查看导致数据倾斜的key的数据分布情况
也可从以下几种情况考虑：
1、是不是有OOM情况出现，一般是少数内存溢出的问题
2、是不是应用运行时间差异很大，总体时间很长
3、需要了解你所处理的数据Key的分布情况，如果有些Key有大量的条数，那么就要小心数据倾斜的问题
4、一般需要通过Spark Web UI和其他一些监控方式出现的异常来综合判断
5、看看代码里面是否有一些导致Shuffle的算子出现
4. 数据倾斜的几种典型情况（重点）
- 数据源中的数据分布不均匀，Spark需要频繁交互
- 数据集中的不同Key由于分区方式，导致数据倾斜
- JOIN操作中，一个数据集中的数据分布不均匀，另一个数据集较小（主要）
- 聚合操作中，数据集中的数据分布不均匀（主要）
- JOIN操作中，两个数据集都比较大，其中只有几个Key的数据分布不均匀
- JOIN操作中，两个数据集都比较大，有很多Key的数据分布不均匀
- 数据集中少数几个key数据量很大，不重要，其他数据均匀
注意：
需要处理的数据倾斜问题就是Shuffle后数据的分布是否均匀问题
只要保证最后的结果是正确的，可以采用任何方式来处理数据倾斜，只要保证在处理过程中不发生数据倾斜就可以
5. 数据倾斜的处理方法
发现数据倾斜的时候，不要急于提高executor的资源，修改参数或是修改程序，首先要检查数据本身，是否存在异常数据。
5.1 检查数据，找出异常的key
如果任务长时间卡在最后1个(几个)任务，首先要对key进行抽样分析，判断是哪些key造成的。
选取key，对数据进行抽样，统计出现的次数，根据出现次数大小排序取出前几个
df.select("key").sample(false,0.1).(k=>(k,1)).reduceBykey(_+_).map(k=>(k._2,k._1)).sortByKey(false).take(10) 如果发现多数数据分布都较为平均，而个别数据比其他数据大上若干个数量级，则说明发生了数据倾斜。
经过分析，倾斜的数据主要有以下三种情况:
null（空值）或是一些无意义的信息()之类的,大多是这个原因引起。
无效数据，大量重复的测试数据或是对结果影响不大的有效数据。
有效数据，业务导致的正常数据分布。
解决办法
第1，2种情况，直接对数据进行过滤即可。
第3种情况则需要进行一些特殊操作，常见的有以下几种做法。
隔离执行，将异常的key过滤出来单独处理，最后与正常数据的处理结果进行union操作。
对key先添加随机值，进行操作后，去掉随机值，再进行一次操作。
使用reduceByKey 代替 groupByKey
使用map join。
举例：
如果使用reduceByKey因为数据倾斜造成运行失败的问题。具体操作如下：将原始的 key 转化为 key + 随机值(例如Random.nextInt)
对数据进行 reduceByKey(func)
将 key + 随机值 转成 key
再对数据进行 reduceByKey(func)
tip1: 如果此时依旧存在问题，建议筛选出倾斜的数据单独处理。最后将这份数据与正常的数据进行union即可。tips2: 单独处理异常数据时，可以配合使用Map Join解决
5.1.1 数据源中的数据分布不均匀，Spark需要频繁交互
解决方案1：避免数据源的数据倾斜
实现原理：通过在Hive中对倾斜的数据进行预处理，以及在进行kafka数据分发时尽量进行平均分配。这种方案从根源上解决了数据倾斜，彻底避免了在Spark中执行shuffle类算子，那么肯定就不会有数据倾斜的问题了。
方案优点：实现起来简单便捷，效果还非常好，完全规避掉了数据倾斜，Spark作业的性能会大幅度提升。
方案缺点：治标不治本，Hive或者Kafka中还是会发生数据倾斜。
适用情况：在一些Java系统与Spark结合使用的项目中，会出现Java代码频繁调用Spark作业的场景，而且对Spark作业的执行性能要求很高，就比较适合使用这种方案。将数据倾斜提前到上游的Hive ETL，每天仅执行一次，只有那一次是比较慢的，而之后每次Java调用Spark作业时，执行速度都会很快，能够提供更好的用户体验。
总结：前台的Java系统和Spark有很频繁的交互，这个时候如果Spark能够在最短的时间内处理数据，往往会给前端有非常好的体验。这个时候可以将数据倾斜的问题抛给数据源端，在数据源端进行数据倾斜的处理。但是这种方案没有真正的处理数据倾斜问题
5.1.2 数据集中的不同Key由于分区方式，导致数据倾斜
解决方案1：调整并行度
实现原理：增加shuffle read task的数量，可以让原本分配给一个task的多个key分配给多个task，从而让每个task处理比原来更少的数据。
方案优点：实现起来比较简单，可以有效缓解和减轻数据倾斜的影响。
方案缺点：只是缓解了数据倾斜而已，没有彻底根除问题，根据实践经验来看，其效果有限。
实践经验：该方案通常无法彻底解决数据倾斜，因为如果出现一些极端情况，比如某个key对应的数据量有100万，那么无论你的task数量增加到多少，都无法处理。
总结：调整并行度：适合于有大量key由于分区算法或者分区数的问题，将key进行了不均匀分区，可以通过调大或者调小分区数来试试是否有效
解决方案2：
缓解数据倾斜**（自定义Partitioner）**
适用场景：大量不同的Key被分配到了相同的Task造成该Task数据量过大。
解决方案： 使用自定义的Partitioner实现类代替默认的HashPartitioner，尽量将所有不同的Key均匀分配到不同的Task中。
优势： 不影响原有的并行度设计。如果改变并行度，后续Stage的并行度也会默认改变，可能会影响后续Stage。
劣势： 适用场景有限，只能将不同Key分散开，对于同一Key对应数据集非常大的场景不适用。效果与调整并行度类似，只能缓解数据倾斜而不能完全消除数据倾斜。而且需要根据数据特点自定义专用的Partitioner，不够灵活。
5.2 检查Spark运行过程相关操作
5.2.1 JOIN操作中，一个数据集中的数据分布不均匀，另一个数据集较小（主要）
解决方案：Reduce side Join转变为Map side Join
方案适用场景：在对RDD使用join类操作，或者是在Spark SQL中使用join语句时，而且join操作中的一个RDD或表的数据量比较小（比如几百M），比较适用此方案。
方案实现原理：普通的join是会走shuffle过程的，而一旦shuffle，就相当于会将相同key的数据拉取到一个shuffle read task中再进行join，此时就是reduce join。但是如果一个RDD是比较小的，则可以采用广播小RDD全量数据+map算子来实现与join同样的效果，也就是map join，此时就不会发生shuffle操作，也就不会发生数据倾斜。
方案优点：对join操作导致的数据倾斜，效果非常好，因为根本就不会发生shuffle，也就根本不会发生数据倾斜。
方案缺点：适用场景较少，因为这个方案只适用于一个大表和一个小表的情况。
5.2.2 聚合操作中，数据集中的数据分布不均匀（主要）
解决方案：两阶段聚合（局部聚合+全局聚合）
适用场景：对RDD执行reduceByKey等聚合类shuffle算子或者在Spark SQL中使用group by语句进行分组聚合时，比较适用这种方案
实现原理：将原本相同的key通过附加随机前缀的方式，变成多个不同的key，就可以让原本被一个task处理的数据分散到多个task上去做局部聚合，进而解决单个task处理数据量过多的问题。接着去除掉随机前缀，再次进行全局聚合，就可以得到最终的结果。具体原理见下图。
优点：对于聚合类的shuffle操作导致的数据倾斜，效果是非常不错的。通常都可以解决掉数据倾斜，或者至少是大幅度缓解数据倾斜，将Spark作业的性能提升数倍以上。
缺点：仅仅适用于聚合类的shuffle操作，适用范围相对较窄。如果是join类的shuffle操作，还得用其他的解决方案
将相同key的数据分拆处理
5.2.3 JOIN操作中，两个数据集都比较大，其中只有几个Key的数据分布不均匀
解决方案：为倾斜key增加随机前/后缀
适用场景：两张表都比较大，无法使用Map侧Join。其中一个RDD有少数几个Key的数据量过大，另外一个RDD的Key分布较为均匀。
解决方案：将有数据倾斜的RDD中倾斜Key对应的数据集单独抽取出来加上随机前缀，另外一个RDD每条数据分别与随机前缀结合形成新的RDD（笛卡尔积，相当于将其数据增到到原来的N倍，N即为随机前缀的总个数），然后将二者Join后去掉前缀。然后将不包含倾斜Key的剩余数据进行Join。最后将两次Join的结果集通过union合并，即可得到全部Join结果。
优势：相对于Map侧Join，更能适应大数据集的Join。如果资源充足，倾斜部分数据集与非倾斜部分数据集可并行进行，效率提升明显。且只针对倾斜部分的数据做数据扩展，增加的资源消耗有限。
劣势：如果倾斜Key非常多，则另一侧数据膨胀非常大，此方案不适用。而且此时对倾斜Key与非倾斜Key分开处理，需要扫描数据集两遍，增加了开销。
注意：具有倾斜Key的RDD数据集中，key的数量比较少
5.2.4 JOIN操作中，两个数据集都比较大，有很多Key的数据分布不均匀
解决方案：随机前缀和扩容RDD进行join
适用场景：如果在进行join操作时，RDD中有大量的key导致数据倾斜，那么进行分拆key也没什么意义。
实现思路：将该RDD的每条数据都打上一个n以内的随机前缀。同时对另外一个正常的RDD进行扩容，将每条数据都扩容成n条数据，扩容出来的每条数据都依次打上一个0~n的前缀。最后将两个处理后的RDD进行join即可。和上一种方案是尽量只对少数倾斜key对应的数据进行特殊处理，由于处理过程需要扩容RDD，因此上一种方案扩容RDD后对内存的占用并不大；而这一种方案是针对有大量倾斜key的情况，没法将部分key拆分出来进行单独处理，因此只能对整个RDD进行数据扩容，对内存资源要求很高。
优点：对join类型的数据倾斜基本都可以处理，而且效果也相对比较显著，性能提升效果非常不错。
缺点：该方案更多的是缓解数据倾斜，而不是彻底避免数据倾斜。而且需要对整个RDD进行扩容，对内存资源要求很高。
实践经验：曾经开发一个数据需求的时候，发现一个join导致了数据倾斜。优化之前，作业的执行时间大约是60分钟左右；使用该方案优化之后，执行时间缩短到10分钟左右，性能提升了6倍。
注意：将倾斜Key添加1-N的随机前缀，并将被Join的数据集相应的扩大N倍（需要将1-N数字添加到每一条数据上作为前缀）
5.2.5 数据集中少数几个key数据量很大，不重要，其他数据均匀
解决方案：过滤少数倾斜Key
适用场景：如果发现导致倾斜的key就少数几个，而且对计算本身的影响并不大的话，那么很适合使用这种方案。比如99%的key就对应10条数据，但是只有一个key对应了100万数据，从而导致了数据倾斜。
优点：实现简单，而且效果也很好，可以完全规避掉数据倾斜。
缺点：适用场景不多，大多数情况下，导致倾斜的key还是很多的，并不是只有少数几个。
实践经验：在项目中我们也采用过这种方案解决数据倾斜。有一次发现某一天Spark作业在运行的时候突然OOM了，追查之后发现，是Hive表中的某一个key在那天数据异常，导致数据量暴增。因此就采取每次执行前先进行采样，计算出样本中数据量最大的几个key之后，直接在程序中将那些key给过滤掉。
