Spark数据倾斜定位及办理方案

一、调优概述
二、数据倾斜发生时的现象
三、数据倾斜发生的原理
四、如何定位导致数据倾斜的代码
五、某个task实验特殊慢的环境
六、某个task莫名其妙内存溢出的环境
七、查看导致数据倾斜的key的数据分布环境
八、数据倾斜的办理方案：

• 办理方案一：使用 Hive ETL 预处理数据
  
• 办理方案二：过滤少数导致倾斜的key
  
• 办理方案三：进步shuffle操作的并行度
  
• 办理方案四：两阶段聚合（局部聚合+全局聚合）
  
• 办理方案五：将reduce join转为map join
  
• 办理方案六：采样倾斜key并分拆join操作
  
• 办理方案七：使用随机前缀和扩容RDD进行join
  
• 办理方案八：多种方案组合使用
  

一、调优概述

有的时间，我们大概会遇到大数据盘算中一个最棘手的题目——数据倾斜，此时Spark作业的性能会比期望差很多。数据倾斜调优，就是使用各种技能方案办理差别类型的数据倾斜题目，以包管Spark作业的性能。
二、数据倾斜发生时的现象

绝大多数task实验得都非常快，但个别task实验极慢。比如，统共有1000个task，997个task都在1分钟之内实验完了，但是剩余两三个task却要一两个小时。这种环境很常见。原本可以大概正常实验的Spark作业，某天忽然报出OOM（内存溢出）异常，观察异常栈，是我们写的业务代码造成的。这种环境比力少见。
三、数据倾斜发生的原理

数据倾斜的原理很简朴：在进行shuffle的时间，必须将各个节点上相同的key拉取到某个节点上的一个task来进行处理，比如按照key进行聚合或join等操作。此时假如某个key对应的数据量特殊大的话，就会发生数据倾斜。比如大部门key对应10条数据，但是个别key却对应了100万条数据，那么大部门task大概就只会分配到10条数据，然后1秒钟就运行完了；但是个别task大概分配到了100万数据，要运行一两个小时。因此，整个Spark作业的运行进度是由运行时间最长的那个task决定的。
因此出现数据倾斜的时间，Spark作业看起来会运行得非常迟钝，乃至大概因为某个task处理的数据量过大导致内存溢出。
下图就是一个很清晰的例子：hello这个key，在三个节点上对应了统共7条数据，这些数据都会被拉取到同一个task中进行处理；而world和you这两个key分别才对应1条数据，所以别的两个task只要分别处理1条数据即可。此时第一个task的运行时间大概是别的两个task的7倍，而整个stage的运行速率也由运行最慢的那个task所决定。
四、如何定位导致数据倾斜的代码

数据倾斜只会发生在shuffle过程中。这里给各人罗列一些常用的而且大概会触发shuffle操作的算子：distinct、groupByKey、reduceByKey、aggregateByKey、join、cogroup、repartition等。出现数据倾斜时，大概就是你的代码中使用了这些算子中的某一个所导致的。
五、某个task实验特殊慢的环境

起首要看的，就是数据倾斜发生在第几个stage中。假如是用yarn-client模式提交，那么当地是直接可以看到log的，可以在log中找到当前运行到了第几个stage，可以看下之前写的Hive数据倾斜：Hive 数据倾斜题目定位排查及办理
假如是用yarn-cluster模式提交，则可以通过Spark Web UI来查看当前运行到了第几个stage。
此外，无论是使用yarn-client模式照旧yarn-cluster模式，我们都可以在Spark Web UI上深入看一下当前这个stage各个task分配的数据量，从而进一步确定是不是task分配的数据不均匀导致了数据倾斜。
比如下图中，倒数第三列显示了每个task的运行时间。显着可以看到，有的task运行特殊快，只须要几秒钟就可以运行完；而有的task运行特殊慢，须要几分钟才气运行完，此时单从运行时间上看就已经可以大概确定发生数据倾斜了。此外，倒数第一列显示了每个task处理的数据量，显着可以看到，运行时间特殊短的task只须要处理几百KB的数据即可，而运行时间特殊长的task须要处理几千KB的数据，处理的数据量差了10倍。此时更加可以大概确定是发生了数据倾斜。
知道数据倾斜发生在哪一个stage之后，接着我们就须要根据stage划分原理，推算出来发生倾斜的那个stage对应代码中的哪一部门，这部门代码中肯定会有一个shuffle类算子。
精准推算stage与代码的对应关系，须要对Spark的源码有深入的理解，这里我们可以先容一个相对简朴实用的推算方法：只要看到Spark代码中出现了一个shuffle类算子大概是Spark SQL的SQL语句中出现了会导致shuffle的语句（比如group by语句），那么就可以判断，以那个地方为界限划分出了前后两个stage。这里我们就以Spark最底子的入门步伐——单词计数来举例，如何用最简朴的方法大致推算出一个stage对应的代码。如下示例，在整个代码中，只有一个reduceByKey是会发生shuffle的算子，因此就可以认为，以这个算子为界限，会划分出前后两个stage。

• stage0，主要是实验从textFile到map操作，以及实验shuffle write操作。shuffle write操作，我们可以简朴理解为对pairs RDD中的数据进行分区操作，每个task处理的数据中，相同的key会写入同一个磁盘文件内。
  
• stage1，主要是实验从reduceByKey到collect操作，stage1的各个task一开始运行，就会起首实验shuffle read操作。实验shuffle read操作的task，会从stage0的各个task地点节点拉取属于自己处理的那些key，然后对同一个key进行全局性的聚合或join等操作，在这里就是对key的value值进行累加。stage1在实验完reduceByKey算子之后，就盘算出了最终的wordCounts RDD，然后会实验collect算子，将所有数据拉取到Driver上，供我们遍历和打印输出。
  

1. val conf = new SparkConf()
2. val sc = new SparkContext(conf)
4. val lines = sc.textFile("hdfs://...")
5. val words = lines.flatMap(_.split(" "))
6. val pairs = words.map((_, 1))
7. val wordCounts = pairs.reduceByKey(_ + _)
9. wordCounts.collect().foreach(println(_))

复制代码

通过对单词计数步伐的分析，渴望可以大概让各人了解最根本的stage划分的原理，以及stage划分后shuffle操作是如何在两个stage的边界处实验的。然后我们就知道如何快速定位出发生数据倾斜的stage对应代码的哪一个部门了。比如我们在Spark Web UI大概当地log中发现，stage1的某几个task实验得特殊慢，判断stage1出现了数据倾斜，那么就可以回到代码中定位出stage1主要包括了reduceByKey这个shuffle类算子，此时根本就可以确定是由educeByKey算子导致的数据倾斜题目。比如某个单词出现了100万次，其他单词才出现10次，那么stage1的某个task就要处理100万数据，整个stage的速率就会被这个task拖慢。
六、某个task莫名其妙内存溢出的环境

这种环境下去定位出题目标代码就比力容易了。我们发起直接看yarn-client模式下当地log的异常栈，大概是通过YARN查看yarn-cluster模式下的log中的异常栈。一样平常来说，通过异常栈信息就可以定位到你的代码中哪一行发生了内存溢出。然后在那行代码附近找找，一样平常也会有shuffle类算子，此时很大概就是这个算子导致了数据倾斜。但是各人要注意的是，不能单纯靠偶尔的内存溢出就判断发生了数据倾斜。因为自己编写的代码的bug，以及偶尔出现的数据异常，也大概会导致内存溢出。因此照旧要按照上面所讲的方法，通过Spark Web UI查看报错的那个stage的各个task的运行时间以及分配的数据量，才气确定是否是由于数据倾斜才导致了这次内存溢出。
七、查看导致数据倾斜的key的数据分布环境

知道了数据倾斜发生在哪里之后，通常须要分析一下那个实验了shuffle操作而且导致了数据倾斜的RDD/Hive表，查看一下其中key的分布环境。这主要是为之后选择哪一种技能方案提供依据。针对差别的key分布与差别的shuffle算子组合起来的各种环境，大概须要选择差别的技能方案来办理。此时根据你实验操作的环境差别，可以有很多种查看key分布的方式：

• 假如是Spark SQL中的group by、join语句导致的数据倾斜，那么就查询一下SQL中使用的表的key分布环境。
  
• 假如是对Spark RDD实验shuffle算子导致的数据倾斜，那么可以在Spark作业中参加查看key分布的代码，比如RDD.countByKey()。然后对统计出来的各个key出现的次数，collect/take到客户端打印一下，就可以看到key的分布环境。
  

举例来说，对于上面所说的单词计数步伐，假如确定了是stage1的reduceByKey算子导致了数据倾斜，那么就应该看看进行reduceByKey操作的RDD中的key分布环境，在这个例子中指的就是pairs RDD。如下示例，我们可以先对pairs采样10%的样本数据，然后使用countByKey算子统计出每个key出现的次数，最后在客户端遍历和打印样本数据中各个key的出现次数。

1. val sampledPairs = pairs.sample(false, 0.1)
2. val sampledWordCounts = sampledPairs.countByKey()
3. sampledWordCounts.foreach(println(_))

复制代码

八、数据倾斜的办理方案

办理方案一：使用 Hive ETL 预处理数据

方案实用场景：导致数据倾斜的是Hive表。假如该Hive表中的数据自己很不均匀（比如某个key对应了100万数据，其他key才对应了10条数据），而且业务场景须要频仍使用Spark对Hive表实验某个分析操作，那么比力适合使用这种技能方案。
方案实现思路：此时可以评估一下，是否可以通过Hive来进行数据预处理（即通过Hive ETL预先对数据按照key进行聚合，大概是预先和其他表进行join），然后在Spark作业中针对的数据源就不是原来的Hive表了，而是预处理后的Hive表。此时由于数据已经预先辈行过聚合或join操作了，那么在Spark作业中也就不须要使用原先的shuffle类算子实验这类操作了。
方案实现原理：这种方案从根源上办理了数据倾斜，因为彻底避免了在Spark中实验shuffle类算子，那么肯定就不会有数据倾斜的题目了。但是这里也要提示一下各人，这种方式属于治标不治本。因为究竟数据自己就存在分布不均匀的题目，所以Hive ETL中进行group by大概join等shuffle操作时，照旧会出现数据倾斜，导致Hive ETL的速率很慢。我们只是把数据倾斜的发生提前到了Hive ETL中，避免Spark步伐发生数据倾斜而已。
方案优点：实现起来简朴便捷，效果还非常好，完全规避掉了数据倾斜，Spark作业的性能会大幅度提升。
方案缺点：治标不治本，Hive ETL中照旧会发生数据倾斜。
方案实践经验：在一些Java体系与Spark结合使用的项目中，会出现Java代码频仍调用Spark作业的场景，而且对Spark作业的实验性能要求很高，就比力适合使用这种方案。将数据倾斜提前到上游的Hive ETL，天天仅实验一次，只有那一次是比力慢的，而之后每次Java调用Spark作业时，实验速率都会很快，可以大概提供更好的用户体验。
项目实践经验：在美团·点评的交互式用户行为分析体系中使用了这种方案，该体系主要是允许用户通过Java Web体系提交数据分析统计任务，后端通过Java提交Spark作业进行数据分析统计。要求Spark作业速率必须要快，尽量在10分钟以内，否则速率太慢，用户体验会很差。所以我们将有些Spark作业的shuffle操作提前到了Hive ETL中，从而让Spark直接使用预处理的Hive中心表，尽大概地镌汰Spark的shuffle操作，大幅度提升了性能，将部门作业的性能提升了6倍以上。
办理方案二：过滤少数导致倾斜的key

方案实用场景：假如发现导致倾斜的key就少数几个，而且对盘算自己的影响并不大的话，那么很适合使用这种方案。比如99%的key就对应10条数据，但是只有一个key对应了100万数据，从而导致了数据倾斜。
方案实现思路：假如我们判断那少数几个数据量特殊多的key，对作业的实验和盘算效果不是特殊重要的话，那么干脆就直接过滤掉那少数几个key。比如，在Spark SQL中可以使用where子句过滤掉这些key大概在Spark Core中对RDD实验filter算子过滤掉这些key。假如须要每次作业实验时，动态判断哪些key的数据量最多然后再进行过滤，那么可以使用sample算子对RDD进行采样，然后盘算出每个key的数目，取数据量最多的key过滤掉即可。
方案实现原理：将导致数据倾斜的key给过滤掉之后，这些key就不会加入盘算了，天然不大概产生数据倾斜。
方案优点：实现简朴，而且效果也很好，可以完全规避掉数据倾斜。
方案缺点：实用场景不多，大多数环境下，导致倾斜的key照旧很多的，并不是只有少数几个。
方案实践经验：在项目中我们也采用过这种方案办理数据倾斜。有一次发现某一天Spark作业在运行的时间忽然OOM了，追查之后发现，是Hive表中的某一个key在那天数据异常，导致数据量暴增。因此就采取每次实验前先辈行采样，盘算出样本中数据量最大的几个key之后，直接在步伐中将那些key给过滤掉。
办理方案三：进步shuffle操作的并行度

方案实用场景：假如我们必须要对数据倾斜迎难而上，那么发起优先使用这种方案，因为这是处理数据倾斜最简朴的一种方案。
方案实现思路：在对RDD实验shuffle算子时，给shuffle算子传入一个参数，比如reduceByKey(1000)，该参数就设置了这个shuffle算子实验时shuffle read task的数目。对于Spark SQL中的shuffle类语句，比如group by、join等，须要设置一个参数，即spark.sql.shuffle.partitions，该参数代表了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的实验时间都会变短了。详细原理如下图所示。
方案优点：实现起来比力简朴，可以有用缓解和减轻数据倾斜的影响。
方案缺点：只是缓解了数据倾斜而已，没有彻底根除题目，根据实践经验来看，其效果有限。
方案实践经验：该方案通常无法彻底办理数据倾斜，因为假如出现一些极度环境，比如某个key对应的数据量有100万，那么无论你的task数目增加到多少，这个对应着100万数据的key肯定照旧会分配到一个task中行止理，因此注定照旧会发生数据倾斜的。所以这种方案只能说是在发现数据倾斜时尝试使用的第一种手段，尝试去用嘴简朴的方法缓解数据倾斜而已，大概是和其他方案结合起来使用。
![][img-2]
办理方案四：两阶段聚合（局部聚合+全局聚合）

方案实用场景：对RDD实验reduceByKey等聚合类shuffle算子大概在Spark SQL中使用group by语句进行分组聚适时，比力实用这种方案。
方案实现思路：这个方案的核心实现思路就是进行两阶段聚合。第一次是局部聚合，先给每个key都打上一个随机数，比如10以内的随机数，此时原先一样的key就变成不一样的了，比如(hello, 1) (hello, 1) (hello, 1) (hello, 1)，就会变成(1_hello, 1) (1_hello, 1) (2_hello, 1) (2_hello, 1)。接着对打上随机数后的数据，实验reduceByKey等聚合操作，进行局部聚合，那么局部聚合效果，就会变成了(1_hello, 2) (2_hello, 2)。然后将各个key的前缀给去掉，就会变成(hello,2)(hello,2)，再次进行全局聚合操作，就可以得到最终效果了，比如(hello, 4)。
方案实现原理：将原本相同的key通过附加随机前缀的方式，变成多个差别的key，就可以让原本被一个task处理的数据分散到多个task上去做局部聚合，进而办理单个task处理数据量过多的题目。接着去撤除随机前缀，再次进行全局聚合，就可以得到最终的效果。详细原理见下图。
方案优点：对于聚合类的shuffle操作导致的数据倾斜，效果好坏常不错的。通常都可以办理掉数据倾斜，大概至少是大幅度缓解数据倾斜，将Spark作业的性能提升数倍以上。
方案缺点：仅仅实用于聚合类的shuffle操作，实用范围相对较窄。假如是join类的shuffle操作，还得用其他的办理方案。
![][img-3]

1. // 第一步，给RDD中的每个key都打上一个随机前缀。
2. JavaPairRDD<String, Long> randomPrefixRdd = rdd.mapToPair(
3.         new PairFunction<Tuple2<Long,Long>, String, Long>() {
4.             private static final long serialVersionUID = 1L;
5.             @Override
6.             public Tuple2<String, Long> call(Tuple2<Long, Long> tuple)
7.                     throws Exception {
8.                 Random random = new Random();
9.                 int prefix = random.nextInt(10);
10.                 return new Tuple2<String, Long>(prefix + "_" + tuple._1, tuple._2);
11.             }
12.         });
14. // 第二步，对打上随机前缀的key进行局部聚合。
15. JavaPairRDD<String, Long> localAggrRdd = randomPrefixRdd.reduceByKey(
16.         new Function2<Long, Long, Long>() {
17.             private static final long serialVersionUID = 1L;
18.             @Override
19.             public Long call(Long v1, Long v2) throws Exception {
20.                 return v1 + v2;
21.             }
22.         });
24. // 第三步，去除RDD中每个key的随机前缀。
25. JavaPairRDD<Long, Long> removedRandomPrefixRdd = localAggrRdd.mapToPair(
26.         new PairFunction<Tuple2<String,Long>, Long, Long>() {
27.             private static final long serialVersionUID = 1L;
28.             @Override
29.             public Tuple2<Long, Long> call(Tuple2<String, Long> tuple)
30.                     throws Exception {
31.                 long originalKey = Long.valueOf(tuple._1.split("_")[1]);
32.                 return new Tuple2<Long, Long>(originalKey, tuple._2);
33.             }
34.         });
36. // 第四步，对去除了随机前缀的RDD进行全局聚合。
37. JavaPairRDD<Long, Long> globalAggrRdd = removedRandomPrefixRdd.reduceByKey(
38.         new Function2<Long, Long, Long>() {
39.             private static final long serialVersionUID = 1L;
40.             @Override
41.             public Long call(Long v1, Long v2) throws Exception {
42.                 return v1 + v2;
43.             }
44.         });

复制代码

办理方案五：将reduce join转为map join

方案实用场景：在对RDD使用join类操作，大概是在Spark SQL中使用join语句时，而且join操作中的一个RDD或表的数据量比力小（比如几百M大概一两G），比力实用此方案。
方案实现思路：不使用join算子进行毗连操作，而使用Broadcast变量与map类算子实现join操作，进而完全规避掉shuffle类的操作，彻底避免数据倾斜的发生和出现。将较小RDD中的数据直接通过collect算子拉取到Driver端的内存中来，然后对其创建一个Broadcast变量；接着对别的一个RDD实验map类算子，在算子函数内，从Broadcast变量中获取较小RDD的全量数据，与当前RDD的每一条数据按照毗连key进行比对，假如毗连key相同的话，那么就将两个RDD的数据用你须要的方式毗连起来。
方案实现原理：普通的join是会走shuffle过程的，而一旦shuffle，就相当于会将相同key的数据拉取到一个shuffle read task中再进行join，此时就是reduce join。但是假如一个RDD是比力小的，则可以采用广播小RDD全量数据+map算子来实现与join同样的效果，也就是map join，此时就不会发生shuffle操作，也就不会发生数据倾斜。详细原理如下图所示。方案优点：对join操作导致的数据倾斜，效果非常好，因为根本就不会发生shuffle，也就根本不会发生数据倾斜。方案缺点：实用场景较少，因为这个方案只实用于一个大表和一个小表的环境。究竟我们须要将小表进行广播，此时会比力斲丧内存资源，driver和每个Executor内存中都会驻留一份小RDD的全量数据。假如我们广播出去的RDD数据比力大，比如10G以上，那么就大概发生内存溢出了。因此并不适合两个都是大表的环境。
![][img-4]

1. // 首先将数据量比较小的RDD的数据，collect到Driver中来。
2. List<Tuple2<Long, Row>> rdd1Data = rdd1.collect()
3. // 然后使用Spark的广播功能，将小RDD的数据转换成广播变量，这样每个Executor就只有一份RDD的数据。
4. // 可以尽可能节省内存空间，并且减少网络传输性能开销。
5. final Broadcast<List<Tuple2<Long, Row>>> rdd1DataBroadcast = sc.broadcast(rdd1Data);
7. // 对另外一个RDD执行map类操作，而不再是join类操作。
8. JavaPairRDD<String, Tuple2<String, Row>> joinedRdd = rdd2.mapToPair(
9.         new PairFunction<Tuple2<Long,String>, String, Tuple2<String, Row>>() {
10.             private static final long serialVersionUID = 1L;
11.             @Override
12.             public Tuple2<String, Tuple2<String, Row>> call(Tuple2<Long, String> tuple)
13.                     throws Exception {
14.                 // 在算子函数中，通过广播变量，获取到本地Executor中的rdd1数据。
15.                 List<Tuple2<Long, Row>> rdd1Data = rdd1DataBroadcast.value();
16.                 // 可以将rdd1的数据转换为一个Map，便于后面进行join操作。
17.                 Map<Long, Row> rdd1DataMap = new HashMap<Long, Row>();
18.                 for(Tuple2<Long, Row> data : rdd1Data) {
19.                     rdd1DataMap.put(data._1, data._2);
20.                 }
21.                 // 获取当前RDD数据的key以及value。
22.                 String key = tuple._1;
23.                 String value = tuple._2;
24.                 // 从rdd1数据Map中，根据key获取到可以join到的数据。
25.                 Row rdd1Value = rdd1DataMap.get(key);
26.                 return new Tuple2<String, String>(key, new Tuple2<String, Row>(value, rdd1Value));
27.             }
28.         });
30. // 这里得提示一下。
31. // 上面的做法，仅仅适用于rdd1中的key没有重复，全部是唯一的场景。
32. // 如果rdd1中有多个相同的key，那么就得用flatMap类的操作，在进行join的时候不能用map，而是得遍历rdd1所有数据进行join。
33. // rdd2中每条数据都可能会返回多条join后的数据。

复制代码

办理方案六：采样倾斜key并分拆join操作

方案实用场景：两个RDD/Hive表进行join的时间，假如数据量都比力大，无法采用“办理方案五”，那么此时可以看一下两个RDD/Hive表中的key分布环境。假如出现数据倾斜，是因为其中某一个RDD/Hive表中的少数几个key的数据量过大，而另一个RDD/Hive表中的所有key都分布比力均匀，那么采用这个办理方案是比力合适的。
方案实现思路：

• 对包含少数几个数据量过大的key的那个RDD，通过sample算子采样出一份样原来，然后统计一下每个key的数目，盘算出来数据量最大的是哪几个key。
  
• 然后将这几个key对应的数据从原来的RDD中拆分出来，形成一个单独的RDD，并给每个key都打上n以内的随机数作为前缀，而不会导致倾斜的大部门key形成别的一个RDD。
  
• 接着将须要join的另一个RDD，也过滤出来那几个倾斜key对应的数据并形成一个单独的RDD，将每条数据膨胀成n条数据，这n条数据都按顺序附加一个0~n的前缀，不会导致倾斜的大部门key也形成别的一个RDD。
  
• 再将附加了随机前缀的独立RDD与另一个膨胀n倍的独立RDD进行join，此时就可以将原先相同的key打散成n份，分散到多个task中去进行join了。
  
• 而别的两个普通的RDD就照常join即可。
  
• 最后将两次join的效果使用union算子合并起来即可，就是最终的join效果。
  

方案实现原理：对于join导致的数据倾斜，假如只是某几个key导致了倾斜，可以将少数几个key分拆成独立RDD，并附加随机前缀打散成n份去进行join，此时这几个key对应的数据就不会集中在少数几个task上，而是分散到多个task进行join了。详细原理见下图。
方案优点：对于join导致的数据倾斜，假如只是某几个key导致了倾斜，采用该方式可以用最有用的方式打散key进行join。而且只须要针对少数倾斜key对应的数据进行扩容n倍，不须要对全量数据进行扩容。避免了占用过多内存。
方案缺点：假如导致倾斜的key特殊多的话，比如成千上万个key都导致数据倾斜，那么这种方式也不适合。
![][img-5]

1. // 首先从包含了少数几个导致数据倾斜key的rdd1中，采样10%的样本数据。
2. JavaPairRDD<Long, String> sampledRDD = rdd1.sample(false, 0.1);
4. // 对样本数据RDD统计出每个key的出现次数，并按出现次数降序排序。
5. // 对降序排序后的数据，取出top 1或者top 100的数据，也就是key最多的前n个数据。
6. // 具体取出多少个数据量最多的key，由大家自己决定，我们这里就取1个作为示范。
7. JavaPairRDD<Long, Long> mappedSampledRDD = sampledRDD.mapToPair(
8.         new PairFunction<Tuple2<Long,String>, Long, Long>() {
9.             private static final long serialVersionUID = 1L;
10.             @Override
11.             public Tuple2<Long, Long> call(Tuple2<Long, String> tuple)
12.                     throws Exception {
13.                 return new Tuple2<Long, Long>(tuple._1, 1L);
14.             }     
15.         });
16. JavaPairRDD<Long, Long> countedSampledRDD = mappedSampledRDD.reduceByKey(
17.         new Function2<Long, Long, Long>() {
18.             private static final long serialVersionUID = 1L;
19.             @Override
20.             public Long call(Long v1, Long v2) throws Exception {
21.                 return v1 + v2;
22.             }
23.         });
24. JavaPairRDD<Long, Long> reversedSampledRDD = countedSampledRDD.mapToPair(
25.         new PairFunction<Tuple2<Long,Long>, Long, Long>() {
26.             private static final long serialVersionUID = 1L;
27.             @Override
28.             public Tuple2<Long, Long> call(Tuple2<Long, Long> tuple)
29.                     throws Exception {
30.                 return new Tuple2<Long, Long>(tuple._2, tuple._1);
31.             }
32.         });
33. final Long skewedUserid = reversedSampledRDD.sortByKey(false).take(1).get(0)._2;
35. // 从rdd1中分拆出导致数据倾斜的key，形成独立的RDD。
36. JavaPairRDD<Long, String> skewedRDD = rdd1.filter(
37.         new Function<Tuple2<Long,String>, Boolean>() {
38.             private static final long serialVersionUID = 1L;
39.             @Override
40.             public Boolean call(Tuple2<Long, String> tuple) throws Exception {
41.                 return tuple._1.equals(skewedUserid);
42.             }
43.         });
44. // 从rdd1中分拆出不导致数据倾斜的普通key，形成独立的RDD。
45. JavaPairRDD<Long, String> commonRDD = rdd1.filter(
46.         new Function<Tuple2<Long,String>, Boolean>() {
47.             private static final long serialVersionUID = 1L;
48.             @Override
49.             public Boolean call(Tuple2<Long, String> tuple) throws Exception {
50.                 return !tuple._1.equals(skewedUserid);
51.             }
52.         });
54. // rdd2，就是那个所有key的分布相对较为均匀的rdd。
55. // 这里将rdd2中，前面获取到的key对应的数据，过滤出来，分拆成单独的rdd，并对rdd中的数据使用flatMap算子都扩容100倍。
56. // 对扩容的每条数据，都打上0～100的前缀。
57. JavaPairRDD<String, Row> skewedRdd2 = rdd2.filter(
58.          new Function<Tuple2<Long,Row>, Boolean>() {
59.             private static final long serialVersionUID = 1L;
60.             @Override
61.             public Boolean call(Tuple2<Long, Row> tuple) throws Exception {
62.                 return tuple._1.equals(skewedUserid);
63.             }
64.         }).flatMapToPair(new PairFlatMapFunction<Tuple2<Long,Row>, String, Row>() {
65.             private static final long serialVersionUID = 1L;
66.             @Override
67.             public Iterable<Tuple2<String, Row>> call(
68.                     Tuple2<Long, Row> tuple) throws Exception {
69.                 Random random = new Random();
70.                 List<Tuple2<String, Row>> list = new ArrayList<Tuple2<String, Row>>();
71.                 for(int i = 0; i < 100; i++) {
72.                     list.add(new Tuple2<String, Row>(i + "_" + tuple._1, tuple._2));
73.                 }
74.                 return list;
75.             }
77.         });
79. // 将rdd1中分拆出来的导致倾斜的key的独立rdd，每条数据都打上100以内的随机前缀。
80. // 然后将这个rdd1中分拆出来的独立rdd，与上面rdd2中分拆出来的独立rdd，进行join。
81. JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD1 = skewedRDD.mapToPair(
82.         new PairFunction<Tuple2<Long,String>, String, String>() {
83.             private static final long serialVersionUID = 1L;
84.             @Override
85.             public Tuple2<String, String> call(Tuple2<Long, String> tuple)
86.                     throws Exception {
87.                 Random random = new Random();
88.                 int prefix = random.nextInt(100);
89.                 return new Tuple2<String, String>(prefix + "_" + tuple._1, tuple._2);
90.             }
91.         })
92.         .join(skewedUserid2infoRDD)
93.         .mapToPair(new PairFunction<Tuple2<String,Tuple2<String,Row>>, Long, Tuple2<String, Row>>() {
94.                         private static final long serialVersionUID = 1L;
95.                         @Override
96.                         public Tuple2<Long, Tuple2<String, Row>> call(
97.                             Tuple2<String, Tuple2<String, Row>> tuple)
98.                             throws Exception {
99.                             long key = Long.valueOf(tuple._1.split("_")[1]);
100.                             return new Tuple2<Long, Tuple2<String, Row>>(key, tuple._2);
101.                         }
102.                     });
104. // 将rdd1中分拆出来的包含普通key的独立rdd，直接与rdd2进行join。
105. JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD2 = commonRDD.join(rdd2);
107. // 将倾斜key join后的结果与普通key join后的结果，uinon起来。
108. // 就是最终的join结果。
109. JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD = joinedRDD1.union(joinedRDD2);

复制代码

办理方案七：使用随机前缀和扩容RDD进行join

方案实用场景：假如在进行join操作时，RDD中有大量的key导致数据倾斜，那么进行分拆key也没什么意义，此时就只能使用最后一种方案来办理题目了。
方案实现思路：

• 该方案的实现思路根本和“办理方案六”类似，起首查看RDD/Hive表中的数据分布环境，找到那个造成数据倾斜的RDD/Hive表，比如有多个key都对应了超过1万条数据。
  
• 然后将该RDD的每条数据都打上一个n以内的随机前缀。
  
• 同时对别的一个正常的RDD进行扩容，将每条数据都扩容成n条数据，扩容出来的每条-数据都依次打上一个0~n的前缀。
  
• 最后将两个处理后的RDD进行join即可。
  

方案实现原理：将原先一样的key通过附加随机前缀变成不一样的key，然后就可以将这些处理后的“差别key”分散到多个task中行止理，而不是让一个task处理大量的相同key。该方案与“办理方案六”的差别之处就在于，上一种方案是尽量只对少数倾斜key对应的数据进行特殊处理，由于处理过程须要扩容RDD，因此上一种方案扩容RDD后对内存的占用并不大；而这一种方案是针对有大量倾斜key的环境，没法将部门key拆分出来进行单独处理，因此只能对整个RDD进行数据扩容，对内存资源要求很高。
方案优点：对join类型的数据倾斜根本都可以处理，而且效果也相对比力显著，性能提升效果非常不错。
方案缺点：该方案更多的是缓解数据倾斜，而不是彻底避免数据倾斜。而且须要对整个RDD进行扩容，对内存资源要求很高。
方案实践经验：曾经开发一个数据需求的时间，发现一个join导致了数据倾斜。优化之前，作业的实验时间约莫是60分钟左右；使用该方案优化之后，实验时间收缩到10分钟左右，性能提升了6倍。

1. // 首先将其中一个key分布相对较为均匀的RDD膨胀100倍。
2. JavaPairRDD<String, Row> expandedRDD = rdd1.flatMapToPair(
3.         new PairFlatMapFunction<Tuple2<Long,Row>, String, Row>() {
4.             private static final long serialVersionUID = 1L;
5.             @Override
6.             public Iterable<Tuple2<String, Row>> call(Tuple2<Long, Row> tuple)
7.                     throws Exception {
8.                 List<Tuple2<String, Row>> list = new ArrayList<Tuple2<String, Row>>();
9.                 for(int i = 0; i < 100; i++) {
10.                     list.add(new Tuple2<String, Row>(0 + "_" + tuple._1, tuple._2));
11.                 }
12.                 return list;
13.             }
14.         });
16. // 其次，将另一个有数据倾斜key的RDD，每条数据都打上100以内的随机前缀。
17. JavaPairRDD<String, String> mappedRDD = rdd2.mapToPair(
18.         new PairFunction<Tuple2<Long,String>, String, String>() {
19.             private static final long serialVersionUID = 1L;
20.             @Override
21.             public Tuple2<String, String> call(Tuple2<Long, String> tuple)
22.                     throws Exception {
23.                 Random random = new Random();
24.                 int prefix = random.nextInt(100);
25.                 return new Tuple2<String, String>(prefix + "_" + tuple._1, tuple._2);
26.             }
27.         });
29. // 将两个处理后的RDD进行join即可。
30. JavaPairRDD<String, Tuple2<String, Row>> joinedRDD = mappedRDD.join(expandedRDD);

复制代码

办理方案八：多种方案组合使用

在实践中发现，很多环境下，假如只是处理较为简朴的数据倾斜场景，那么使用上述方案中的某一种根本就可以办理。但是假如要处理一个较为复杂的数据倾斜场景，那么大概须要将多种方案组合起来使用。比如说，我们针对出现了多个数据倾斜环节的Spark作业，可以先运用办理方案一和二，预处理一部门数据，并过滤一部门数据来缓解；其次可以对某些shuffle操作提升并行度，优化其性能；最后还可以针对差别的聚合或join操作，选择一种方案来优化其性能。各人须要对这些方案的思路和原理都透彻理解之后，在实践中根据各种差别的环境，灵活运用多种方案，来办理自己的数据倾斜题目。
参考文档：

• 最强最全面的数仓建设规范指南
  
• 美团数据平台及数仓建设实践，超十万字总结
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。