关于处理倾斜的demo程序-githup

解决方案四：两阶段聚合（局部聚合+全局聚合）

• 方案适用场景：对RDD执行reduceByKey等聚合类shuffle算子或者在Spark SQL中使用group by语句进行分组聚合时，比较适用这种方案。

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

• 方案优点：对于聚合类的shuffle操作导致的数据倾斜，效果是非常不错的。通常都可以解决掉数据倾斜，或者至少是大幅度缓解数据倾斜，将Spark作业的性能提升数倍以上。

• 方案缺点：仅仅适用于聚合类的shuffle操作，适用范围相对较窄。如果是join类的shuffle操作，还得用其他的解决方案。

    public static void main(String[] args) {
        SparkSession spark = SparkSession
                .builder()
                .appName("JavaWordCount")
                .master("local[*]")
                .getOrCreate();

        JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());

        int slices = 3;
        List<String> list1 = new ArrayList(Arrays.asList("hello 1",
                "hello 2",
                "hello 3",
                "hello 4",
                "you 1",
                "me 2"));

        JavaPairRDD<String, Long> rdd = jsc.parallelize(list1, slices).mapToPair(
                new PairFunction<String, String, Long>() {

                    @Override
                    public Tuple2<String, Long> call(String s) throws Exception {
                        String[] arr = s.split(" ");

                        return new Tuple2<String, Long>(arr[0], Long.parseLong(arr[1]));
                    }
                }
        );
        // 第一步，给RDD中的每个key都打上一个随机前缀。
        JavaPairRDD<String, Long> randomPrefixRdd = rdd.mapToPair(
                new PairFunction<Tuple2<String, Long>, String, Long>() {
                    @Override
                    public Tuple2<String, Long> call(Tuple2<String, Long> tuple) throws Exception {
                        Random random = new Random();
                        int prefix = random.nextInt(10);
                        return new Tuple2<String, Long>(prefix + "_" + tuple._1, tuple._2);
                    }
                }
        );

        // 第二步，对打上随机前缀的key进行局部聚合。
        JavaPairRDD<String, Long> localAggrdd = randomPrefixRdd.reduceByKey(
                new Function2<Long, Long, Long>() {
                    private static final long serialVersionUID = 1L;

                    @Override
                    public Long call(Long v1, Long v2) throws Exception {
                        return v1 + v2;
                    }
                });

// 第三步，去除RDD中每个key的随机前缀。
        JavaPairRDD<String, Long> removedandomPrefixdd = localAggrdd.mapToPair(
                new PairFunction<Tuple2<String, Long>, String, Long>() {
                    private static final long serialVersionUID = 1L;

                    @Override
                    public Tuple2<String, Long> call(Tuple2<String, Long> tuple)
                            throws Exception {
                        String originalKey = tuple._1.split("_")[1];
                        return new Tuple2<String, Long>(originalKey, tuple._2);
                    }
                });

// 第四步，对去除了随机前缀的RDD进行全局聚合。
        JavaPairRDD<String, Long> globalAggrdd = removedandomPrefixdd.reduceByKey(
                new Function2<Long, Long, Long>() {
                    private static final long serialVersionUID = 1L;

                    @Override
                    public Long call(Long v1, Long v2) throws Exception {
                        return v1 + v2;
                    }
                });

        List<Tuple2<String, Long>> output = globalAggrdd.collect();
        for (Tuple2<?, ?> tuple : output) {
            System.out.println(tuple._1() + ": " + tuple._2());
        }
        spark.stop();

    }

解决方案五：将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以上，那么就可能发生内存溢出了。因此并不适合两个都是大表的情况。

  
  image.png

 SparkSession spark = SparkSession
                .builder()
                .appName("JavaWordCount")
                .master("local[*]")
                .getOrCreate();

        JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
        SparkContext sc = spark.sparkContext();

        int slices = 3;
        List<String> list1 = new ArrayList(Arrays.asList(
                "1  郑祥楷1",
                "2  王佳豪1",
                "3  刘鹰2",
                "4  宋志华3",
                "5  刘帆4",
                "6  OLDLi5"
        ));

        List<String> list2 = new ArrayList(Arrays.asList(
                "1 1807bd-bj",
                "2 1807bd-sz",
                "3 1807bd-wh",
                "4 1807bd-xa",
                "7 1805bd-bj"
        ));

        JavaPairRDD<Long, String> rdd1 = jsc.parallelize(list1, slices).mapToPair(
                new PairFunction<String, Long, String>() {

                    @Override
                    public Tuple2<Long, String> call(String s) throws Exception {
                        String[] arr = s.split("  ");

                        return new Tuple2<Long, String>(Long.parseLong(arr[0]), arr[1]);
                    }
                }
        );

        JavaPairRDD<Long, String> rdd2 = jsc.parallelize(list2, slices).mapToPair(
                new PairFunction<String, Long, String>() {

                    @Override
                    public Tuple2<Long, String> call(String s) throws Exception {
                        String[] arr = s.split(" ");

                        return new Tuple2<Long, String>(Long.parseLong(arr[0]), arr[1]);
                    }
                }
        );
        // 首先将数据量比较小的RDD的数据，collect到Driver中来。
        List<Tuple2<Long, String>> rdd1Data = rdd1.collect();
        final Broadcast<List<Tuple2<Long, String>>> rdd1DataBroadcast = jsc.broadcast(rdd1Data);
        // 然后使用Spark的广播功能，将小RDD的数据转换成广播变量，这样每个Executor就只有一份RDD的数据。
        // 可以尽可能节省内存空间，并且减少网络传输性能开销。
        // 对另外一个RDD执行map类操作，而不再是join类操作。
        JavaPairRDD<Long, Tuple2<String, String>> joinedRdd = rdd2.mapToPair(
                new PairFunction<Tuple2<Long, String>, Long, Tuple2<String, String>>() {
                    private static final long serialVersionUID = 1L;

                    @Override
                    public Tuple2<Long, Tuple2<String, String>> call(Tuple2<Long, String> tuple)
                            throws Exception {
                        // 在算子函数中，通过广播变量，获取到本地Executor中的rdd1数据。
                        List<Tuple2<Long, String>> rdd1Data = rdd1DataBroadcast.value();
                        // 可以将rdd1的数据转换为一个Map，便于后面进行join操作。
                        Map<Long, String> rdd1DataMap = new HashMap<Long, String>();
                        for (Tuple2<Long, String> data : rdd1Data) {
                            rdd1DataMap.put(data._1, data._2);
                        }
                        // 获取当前RDD数据的key以及value。
                        Long key = tuple._1;
                        String value = tuple._2;
                        // 从rdd1数据Map中，根据key获取到可以join到的数据。
                        String rdd1Value = rdd1DataMap.get(key);
                        return new Tuple2<Long, Tuple2<String, String>>(key, new Tuple2<String, String>(value, rdd1Value));
                    }
                });

        List<Tuple2<Long, Tuple2<String, String>>> output = joinedRdd.collect();
        for (Tuple2<?, ?> tuple : output) {
            Tuple2<String, String> tuple2 = (Tuple2<String, String>)tuple._2();
            System.out.println(tuple._1() + ": " + tuple2._1+ ": " +tuple2._2());
        }
        spark.stop();

解决方案六：采样倾斜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都导致数据倾斜，那么这种方式也不适合。

  
  image.png

// 首先从包含了少数几个导致数据倾斜key的rdd1中，采样10%的样本数据。
JavaPairRDD<Long, String> sampledRDD = rdd1.sample(false, 0.1);
  
// 对样本数据RDD统计出每个key的出现次数，并按出现次数降序排序。
// 对降序排序后的数据，取出top 1或者top 100的数据，也就是key最多的前n个数据。
// 具体取出多少个数据量最多的key，由大家自己决定，我们这里就取1个作为示范。
JavaPairRDD<Long, Long> mappedSampledRDD = sampledRDD.mapToPair(
        new PairFunction<Tuple2<Long,String>, Long, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<Long, Long> call(Tuple2<Long, String> tuple)
                    throws Exception {
                return new Tuple2<Long, Long>(tuple._1, 1L);
            }     
        });
JavaPairRDD<Long, Long> countedSampledRDD = mappedSampledRDD.reduceByKey(
        new Function2<Long, Long, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Long call(Long v1, Long v2) throws Exception {
                return v1 + v2;
            }
        });
JavaPairRDD<Long, Long> reversedSampledRDD = countedSampledRDD.mapToPair( 
        new PairFunction<Tuple2<Long,Long>, Long, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<Long, Long> call(Tuple2<Long, Long> tuple)
                    throws Exception {
                return new Tuple2<Long, Long>(tuple._2, tuple._1);
            }
        });
final Long skewedUserid = reversedSampledRDD.sortByKey(false).take(1).get(0)._2;
  
// 从rdd1中分拆出导致数据倾斜的key，形成独立的RDD。
JavaPairRDD<Long, String> skewedRDD = rdd1.filter(
        new Function<Tuple2<Long,String>, Boolean>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Boolean call(Tuple2<Long, String> tuple) throws Exception {
                return tuple._1.equals(skewedUserid);
            }
        });
// 从rdd1中分拆出不导致数据倾斜的普通key，形成独立的RDD。
JavaPairRDD<Long, String> commonRDD = rdd1.filter(
        new Function<Tuple2<Long,String>, Boolean>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Boolean call(Tuple2<Long, String> tuple) throws Exception {
                return !tuple._1.equals(skewedUserid);
            } 
        });
  
// rdd2，就是那个所有key的分布相对较为均匀的rdd。
// 这里将rdd2中，前面获取到的key对应的数据，过滤出来，分拆成单独的rdd，并对rdd中的数据使用flatMap算子都扩容100倍。
// 对扩容的每条数据，都打上0～100的前缀。
JavaPairRDD<String, Row> skewedRdd2 = rdd2.filter(
         new Function<Tuple2<Long,Row>, Boolean>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Boolean call(Tuple2<Long, Row> tuple) throws Exception {
                return tuple._1.equals(skewedUserid);
            }
        }).flatMapToPair(new PairFlatMapFunction<Tuple2<Long,Row>, String, Row>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Iterable<Tuple2<String, Row>> call(
                    Tuple2<Long, Row> tuple) throws Exception {
                Random random = new Random();
                List<Tuple2<String, Row>> list = new ArrayList<Tuple2<String, Row>>();
                for(int i = 0; i < 100; i++) {
                    list.add(new Tuple2<String, Row>(i + "_" + tuple._1, tuple._2));
                }
                return list;
            }
              
        });
 
// 将rdd1中分拆出来的导致倾斜的key的独立rdd，每条数据都打上100以内的随机前缀。
// 然后将这个rdd1中分拆出来的独立rdd，与上面rdd2中分拆出来的独立rdd，进行join。
JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD1 = skewedRDD.mapToPair(
        new PairFunction<Tuple2<Long,String>, String, String>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<String, String> call(Tuple2<Long, String> tuple)
                    throws Exception {
                Random random = new Random();
                int prefix = random.nextInt(100);
                return new Tuple2<String, String>(prefix + "_" + tuple._1, tuple._2);
            }
        })
        .join(skewedUserid2infoRDD)
        .mapToPair(new PairFunction<Tuple2<String,Tuple2<String,Row>>, Long, Tuple2<String, Row>>() {
                        private static final long serialVersionUID = 1L;
                        @Override
                        public Tuple2<Long, Tuple2<String, Row>> call(
                            Tuple2<String, Tuple2<String, Row>> tuple)
                            throws Exception {
                            long key = Long.valueOf(tuple._1.split("_")[1]);
                            return new Tuple2<Long, Tuple2<String, Row>>(key, tuple._2);
                        }
                    });
 
// 将rdd1中分拆出来的包含普通key的独立rdd，直接与rdd2进行join。
JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD2 = commonRDD.join(rdd2);
 
// 将倾斜key join后的结果与普通key join后的结果，uinon起来。
// 就是最终的join结果。
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进行扩容，对内存资源要求很高

SparkSession spark = SparkSession
                .builder()
                .appName("JavaWordCount")
                .master("local[*]")
                .getOrCreate();

        JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
        SparkContext sc = spark.sparkContext();

        int slices = 3;
        List<String> list2 = new ArrayList(Arrays.asList(
                "1  郑祥楷1",
                "1  郑祥楷1",
                "1  郑祥楷1",
                "1  郑祥楷1",
                "1  郑祥楷1",
                "1  郑祥楷1",
                "1  郑祥楷1",
                "1  郑祥楷1",
                "1  郑祥楷1",
                "1  郑祥楷1",
                "1  郑祥楷1",
                "1  郑祥楷1",
                "2  王佳豪1",
                "2  王佳豪1",
                "2  王佳豪1",
                "2  王佳豪1",
                "2  王佳豪1",
                "2  王佳豪1",
                "2  王佳豪1",
                "2  王佳豪1",
                "2  王佳豪1",
                "3  刘鹰2",
                "4  宋志华3",
                "5  刘帆4",
                "6  OLDLi5"
        ));

        List<String> list1 = new ArrayList(Arrays.asList(
                "1 1807bd-bj",
                "2 1807bd-sz",
                "3 1807bd-wh",
                "4 1807bd-xa",
                "7 1805bd-bj"
        ));

        JavaPairRDD<Long, String> rdd1 = jsc.parallelize(list1, slices).mapToPair(
                new PairFunction<String, Long, String>() {

                    @Override
                    public Tuple2<Long, String> call(String s) throws Exception {
                        String[] arr = s.split(" ");

                        return new Tuple2<Long, String>(Long.parseLong(arr[0]), arr[1]);
                    }
                }
        );

        JavaPairRDD<Long, String> rdd2 = jsc.parallelize(list2, slices).mapToPair(
                new PairFunction<String, Long, String>() {

                    @Override
                    public Tuple2<Long, String> call(String s) throws Exception {
                        String[] arr = s.split("  ");

                        return new Tuple2<Long, String>(Long.parseLong(arr[0]), arr[1]);
                    }
                }
        );

        // 首先将其中一个key分布相对较为均匀的RDD膨胀100倍。
        JavaPairRDD<String, String> expandedRDD = rdd1.flatMapToPair(
                new PairFlatMapFunction<Tuple2<Long, String>, String, String>() {
                    private static final long serialVersionUID = 1L;

                    @Override
                    public Iterator<Tuple2<String, String>> call(Tuple2<Long, String> tuple)
                            throws Exception {
                        List<Tuple2<String, String>> list = new ArrayList<Tuple2<String, String>>();
                        for (int i = 0; i < 100; i++) {
                            list.add(new Tuple2<String, String>(i + "_" + tuple._1, tuple._2));
                        }
                        return list.iterator();
                    }
                });

        List list11 = expandedRDD.collect();


        // 其次，将另一个有数据倾斜key的RDD，每条数据都打上100以内的随机前缀。
        JavaPairRDD<String, String> mappedRDD = rdd2.mapToPair(
                new PairFunction<Tuple2<Long, String>, String, String>() {
                    private static final long serialVersionUID = 1L;

                    @Override
                    public Tuple2<String, String> call(Tuple2<Long, String> tuple)
                            throws Exception {
                        Random random = new Random();
                        int prefix = random.nextInt(100);
                        return new Tuple2<String, String>(prefix + "_" + tuple._1, tuple._2);
                    }
                });

        List list12 = mappedRDD.collect();


        // 将两个处理后的RDD进行join即可。
        JavaPairRDD<String, Tuple2<String, String>> joinedRDD = mappedRDD.join(expandedRDD);

        JavaPairRDD<String, Tuple2<String, String>> returltRDD = joinedRDD.mapToPair(
                new PairFunction<Tuple2<String, Tuple2<String, String>>, String, Tuple2<String, String>>() {
                    @Override
                    public Tuple2<String, Tuple2<String, String>> call(Tuple2<String, Tuple2<String, String>> tuple) throws Exception {
                        String[] arr = tuple._1.split("_");
                        return new Tuple2<String, Tuple2<String, String>>(arr[1], tuple._2);
                    }
                }
        );


        List<Tuple2<String, Tuple2<String, String>>> output = returltRDD.collect();
        for (Tuple2<?, ?> tuple : output) {
            Tuple2<String, String> tuple2 = (Tuple2<String, String>) tuple._2();
            System.out.println(tuple._1() + ": " + tuple2._1 + ": " + tuple2._2());
        }
        spark.stop();

转自:
https://tech.meituan.com/2016/05/12/spark-tuning-pro.html
https://blog.51cto.com/14048416/2338651