Spark数据倾斜解决方案

1、什么是数据倾斜？

       Spark 中的数据倾斜问题主要指shuffle过程中出现的数据倾斜问题，是由于不同的key对应的数据量不同导致的不同task所处理的数据量不同的问题。例如，reduce点一共要处理100万条数据，第一个和第二个task分别被分配到了1万条数据，计算5分钟内完成，第三个task分配到了98万数据，此时第三个task可能需要10个小时完成，这使得整个Spark作业需要10个小时才能运行完成，这就是数据倾斜所带来的后果。

注意，要区分开数据倾斜与数据量过量这两种情况，数据倾斜是指少数task被分配了绝大多数的数据，因此少数task运行缓慢；数据过量是指所有task被分配的数据量都很大，相差不多，所有task都运行缓慢。

2、数据倾斜的表现

1.Spark作业大部分task都执行迅速，但有一个或多个task执行的非常慢，此时就可能出现了数据倾斜，作业可以运行，但是运行的非常缓慢。

2.Spark作业大部分task都执行迅速，但有的task在运行过程中会突然报出OOM，反复执行几次都在某一个task报出OOM错误，此时可能出现数据倾斜，作业无法正常运行。

如图：

3、定位数据倾斜问题

1. 查阅代码中的shuffle算子，例如reduceByKey、countByKey、groupByKey、join等算子，根据代码逻辑判断此处是否会出现数据倾斜。

2.查看 Spark 作业的 log 文件，log 文件对于错误的记录会精确到代码的某一行，可以根据异常定位到的代码位置来明确错误发生在第几个stage，对应的 shuffle 算子是哪一个。

4、解决数据倾斜的方案

1.聚合原数据

1.1避免shuffle过程

      绝大多数情况下，Spark作业的数据来源是Hive表，而入Hive表的数据基本都是经过ETL之后的数据，为了避免数据倾斜发生，我们可以考虑避免shuffle过程，那么就从根本上消除了发生数据倾斜问题的可能。如果Spark作业的数据来源于Hive表，那么可以先在 Hive 表中对数据进行聚合，例如按照 key 进行分组，将同一key 对应的所有value用一种特殊的格式拼接到一个字符串里去，这样，一个key就只有一条数据了；之后，对一个key的所有value进行处理时，只需要进行map操作即可，无需再进行任何的shuffle操作。通过上述方式就避免了执行shuffle操作，也就不可能会发生任何的数据倾斜问题，一定要取分开，处理的数据量打和数据倾斜的区别。

1.2缩小key粒度（增大数据倾斜的可能性，降低每个task的数据量）

key的数量增加，可能会使数据倾斜更加严重

1.3增大key粒度（减小数据倾斜的可能性，增大没个task的数据量）

        如果没有办法对每个key聚合出来一条数据，在特定场景下可以考虑扩大key的聚合粒度。例如当前key的粒度是省，市，区，时间，现在我们考虑扩大粒度，将key的粒度扩大为省，市，时间，这样的话key的数据量会减少，key之间的数据量差异也会减少，由此可以减轻数据倾斜的现象。（此方法的局限性在于应用场景，当应用场景不合适时会加重数据倾斜）

2、过滤导致倾斜的key

1.如果在Spark作业中允许丢弃某些数据，那么可以考虑将可能导致数据倾斜的key进行过滤，过滤可能导致数据倾斜的key对应的数据，这样在Spark作业中就不会发生数据倾斜了。

3、提高shuffle操作中中额度reduce并行度

1.1当方案一和方案二对于数据倾斜的处理没有很好的效果时，可以考虑提高shuffle过程中的reduce端并行度，reduce端并行度的提高就增加了reduce端task的数量，那么每个task分配到的数据量就会相应的减少，由此可以缓解数据倾斜问题。

       reduce端并行度的设置，比如reduceByKey(500)，这个参数会决定shuffle过程中reduce端的并行度，在进行shuffle操作的时候，就会对应着创建指定数量的reduce task。还可以使用         --conf spark.default.parallelism=1000 \来指定reduce端并行度。

       对于SparkSQL中的shuffle类语句，比如group by、join等，可以设置一个参数，即--conf spark.sql.shuffle.partitions=1000 \，该参数代表了shuffle read task的并行度，该默认值是200，对于很多场景来说都有点过小。

       增加shuffle read task的数量，可以让原本分配给一个task的多个key分配给多个task，从而让每个task处理比原来更少的数据。举例来说，如果原本有5个key，每个key对应10条数据，这5个key都是分配给一个task的，那么这个task就要处理50条数据。而增加了shuffle read task以后，每个task就分配到一个key，即每个task就处理10条数据，那么自然每个task的执行时间都会变短了。

1.2reduce端并行度设置存在的缺陷

      提高reduce端并行度并没有从根本上改变数据倾斜的本质和问题（方案一和方案二从根本上避免了数据倾斜的发生）只是尽可能的去缓解和减轻shuffle reduce task的数据压力，以及数据倾斜的问题，适用于有较多key对应的数据量都比较大的情况。改方案通常无法彻底解决数据倾斜，因为一旦出现一些极端情况，比如某个key对应的数据量有100万，那么无论你的task数量增加到多少，这个对应着100万的数据的key肯定还是会被分配到一个task中去处理，因此注定还是会发生数据倾斜。所以这种方案只能说是在发现数据倾斜是尝试使用的第一种手段，尝试去用最简单的方法缓解数据倾斜而已或者是和其他方案结合起来使用。

      在理想情况下，reduce断并行度提升后，会在一定程度上减轻数据倾斜的问题，甚至基本消除数据倾斜，但是在一些情况下，只会让原来由于数据倾斜而运行缓慢的task运行速度稍有提升，或者避免了某些task的OOM问题，但是仍然运行缓慢，此时要及时放弃方案三开始尝试后面的方案。

4、使用随机key实现双重聚合

1.当使用了类似于groupByKey、reduceByKey这样的算子时可以考虑使用随机key实现双重聚合。

       首先，通过map算子给每个数据的key添加随机数前缀，对key进行打散，将原先一样的key变成不一样不一样额度key，然后进行第一次聚合，这样就可以让原本被一个task处理的数据分散到多个task上做局部聚合；

        其次，去掉每个key的前缀，再次进行聚合。

        此方法对于由于groupByKey、reduceByKey这类算子造成的数据倾斜有比较好的效果，仅仅适用于聚合类的shuffle操作，适用范围相对较小。如果是join类的shuffle操作还的使用其它的解决方案。当前三种方案都mei有比较好的效果时要尝试的解决方案。

5、将reduce join转换为map join

       正常情况下，join操作都会执行shuffle过程，并且执行的是reduce join也就是先将所有相同的key和对应的value汇聚到一个reduce task中，然后再进行join操作。普通join的过程如下：

       普通的join是会走shuffle的过程，而一旦shuffle，就相当于会将相同的key的数据拉取到一个shuffle read taskzho那个在进行join，此时就是reduce join。但是如果一个RDD是比较小的，则可以采用广播小RDD全量数据+map算子来实现与join相同的效果，也就是map join，此时就不会发生shuffle操作了，也就就不会发生数据倾斜。

核心思想：不使用join算子进行连接操作，而使用Broadcast变量与map类算子实现join操作进而完全规避掉shuffle类的操作，彻底避免数据倾斜的发生和出现。将较小的RDD中的数据直接通过collect算子拉取到driver端的内存中保存起来，然后对其创建一个Broadcast变量。接着对另一个RDD执行map类算子，在算子函数内，从Broadcast变量中获取较小的RDD的全量数据，与当前RDD的每一条数据按照连接key进行比对，如果连接key相同的话，那么就将两个RDD的数据用你需要的方式连接起来。上述思路根本不会发生shuffle操作从根本上杜绝了join操作可能导致的数据倾斜问题。当join操作有数据倾斜问题并且其中一个RDD的数据量较小时，可以优先考虑这种方式。map join的过程如图：

不适用场景：由于Spark的广播变量是在每个Executor中保存一个副本，如果两个RDD数据量都比较大，那么如果将一个数据量比较大的RDD做成广播变量，那么很有可能会造成内存溢出。

6、sample采样对倾斜key单独进行join

       在Spark中如果某个RDD只有一个key那么在shuffle过程中会默认将此key对应的数据打撒，由不同的reduce断task进行处理。所以，当由单个key导致数据倾斜时，可以将发生数据倾斜的可以单独提取出来，组成一个RDD，然后用这个原本会导致倾斜的key组成的RDD跟其它RDD单独join，此时，根据Spark的运行机制此RDD中的数据会在shuffle阶段被分散到多个task中去进行join操作。

适用场景：对于RDD中的数据，可以将其转换为一个中间表或者是直接使用countByKey的方式，看这个RDD中各个key对应的数据量，此时如果你发现整个RDD就一个key的数据量非常多，那么就可以考虑这种方法。

       当数据量非常大时，可以考虑使用sample采样获取10%的数据，然后分析这10%的数据中哪个key可能会导致数据倾斜，然后将这个key对应的数据单独提取出来。

不适用场景：如果一个RDD中导致数据倾斜的key很多，那么此方案不适用。

7、使用随机数以及扩容进行join

        如果在进行join操作时，RDD中有大量的key导致数据倾斜，那么进行分拆key也没有什么意义，此时就只能使用最后一种方案来解决问题，对于join操作我们可以考虑对其中一个RDD数据进行扩容，另一个RDD进行稀释后再join。

       我们会将原先一样的key通过附加随机前缀变成不一样的key，然后就可以将这些处理后的“不同key“分散到多个task中去处理，而不是让一个task处理大量的相同key。

这种方案是针对有大量倾斜key的情况，没法将部分key拆分出来进行单独处理，需要对整个RDD进行数据扩容，对资源内存要求很高。

核心思想：选择一个RDD，使用flatMap进行扩容，对每条数据的key添加数值前缀，将一条数据映射为多条数据（扩容）。选择另外一个RDD，进行map映射操作，每条数据的key都打上一个随机数作为前缀（稀释）

局限性：如果两个RDD都很大，那么将RDD进行N倍的扩容显然行不通，使用扩容的方式只能缓解数据倾斜，不能彻底解决数据倾斜的问题。