Spark性能優(yōu)化指南——高級(jí)篇

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

通過(guò)對(duì)單詞計(jì)數(shù)程序的分析，希望能夠讓大家了解最基本的stage劃分的原理，以及stage劃分后shuffle操作是如何在兩個(gè)stage的邊界處執(zhí)行的。然后我們就知道如何快速定位出發(fā)生數(shù)據(jù)傾斜的stage對(duì)應(yīng)代碼的哪一個(gè)部分了。比如我們?cè)赟park Web UI或者本地log中發(fā)現(xiàn)，stage1的某幾個(gè)task執(zhí)行得特別慢，判定stage1出現(xiàn)了數(shù)據(jù)傾斜，那么就可以回到代碼中定位出stage1主要包括了reduceByKey這個(gè)shuffle類算子，此時(shí)基本就可以確定是由educeByKey算子導(dǎo)致的數(shù)據(jù)傾斜問(wèn)題。比如某個(gè)單詞出現(xiàn)了100萬(wàn)次，其他單詞才出現(xiàn)10次，那么stage1的某個(gè)task就要處理100萬(wàn)數(shù)據(jù)，整個(gè)stage的速度就會(huì)被這個(gè)task拖慢。

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

數(shù)據(jù)傾斜的解決方案

解決方案一：使用Hive ETL預(yù)處理數(shù)據(jù)

方案適用場(chǎng)景：導(dǎo)致數(shù)據(jù)傾斜的是Hive表。如果該Hive表中的數(shù)據(jù)本身很不均勻（比如某個(gè)key對(duì)應(yīng)了100萬(wàn)數(shù)據(jù)，其他key才對(duì)應(yīng)了10條數(shù)據(jù)），而且業(yè)務(wù)場(chǎng)景需要頻繁使用Spark對(duì)Hive表執(zhí)行某個(gè)分析操作，那么比較適合使用這種技術(shù)方案。

方案實(shí)現(xiàn)思路：此時(shí)可以評(píng)估一下，是否可以通過(guò)Hive來(lái)進(jìn)行數(shù)據(jù)預(yù)處理（即通過(guò)Hive ETL預(yù)先對(duì)數(shù)據(jù)按照key進(jìn)行聚合，或者是預(yù)先和其他表進(jìn)行join），然后在Spark作業(yè)中針對(duì)的數(shù)據(jù)源就不是原來(lái)的Hive表了，而是預(yù)處理后的Hive表。此時(shí)由于數(shù)據(jù)已經(jīng)預(yù)先進(jìn)行過(guò)聚合或join操作了，那么在Spark作業(yè)中也就不需要使用原先的shuffle類算子執(zhí)行這類操作了。

方案實(shí)現(xiàn)原理：這種方案從根源上解決了數(shù)據(jù)傾斜，因?yàn)閺氐妆苊饬嗽赟park中執(zhí)行shuffle類算子，那么肯定就不會(huì)有數(shù)據(jù)傾斜的問(wèn)題了。但是這里也要提醒一下大家，這種方式屬于治標(biāo)不治本。因?yàn)楫吘箶?shù)據(jù)本身就存在分布不均勻的問(wèn)題，所以Hive ETL中進(jìn)行g(shù)roup by或者join等shuffle操作時(shí)，還是會(huì)出現(xiàn)數(shù)據(jù)傾斜，導(dǎo)致Hive ETL的速度很慢。我們只是把數(shù)據(jù)傾斜的發(fā)生提前到了Hive ETL中，避免Spark程序發(fā)生數(shù)據(jù)傾斜而已。

方案優(yōu)點(diǎn)：實(shí)現(xiàn)起來(lái)簡(jiǎn)單便捷，效果還非常好，完全規(guī)避掉了數(shù)據(jù)傾斜，Spark作業(yè)的性能會(huì)大幅度提升。

方案缺點(diǎn)：治標(biāo)不治本，Hive ETL中還是會(huì)發(fā)生數(shù)據(jù)傾斜。

方案實(shí)踐經(jīng)驗(yàn)：在一些Java系統(tǒng)與Spark結(jié)合使用的項(xiàng)目中，會(huì)出現(xiàn)Java代碼頻繁調(diào)用Spark作業(yè)的場(chǎng)景，而且對(duì)Spark作業(yè)的執(zhí)行性能要求很高，就比較適合使用這種方案。將數(shù)據(jù)傾斜提前到上游的Hive ETL，每天僅執(zhí)行一次，只有那一次是比較慢的，而之后每次Java調(diào)用Spark作業(yè)時(shí)，執(zhí)行速度都會(huì)很快，能夠提供更好的用戶體驗(yàn)。

項(xiàng)目實(shí)踐經(jīng)驗(yàn)：在美團(tuán)·點(diǎn)評(píng)的交互式用戶行為分析系統(tǒng)中使用了這種方案，該系統(tǒng)主要是允許用戶通過(guò)Java Web系統(tǒng)提交數(shù)據(jù)分析統(tǒng)計(jì)任務(wù)，后端通過(guò)Java提交Spark作業(yè)進(jìn)行數(shù)據(jù)分析統(tǒng)計(jì)。要求Spark作業(yè)速度必須要快，盡量在10分鐘以內(nèi)，否則速度太慢，用戶體驗(yàn)會(huì)很差。所以我們將有些Spark作業(yè)的shuffle操作提前到了Hive ETL中，從而讓Spark直接使用預(yù)處理的Hive中間表，盡可能地減少Spark的shuffle操作，大幅度提升了性能，將部分作業(yè)的性能提升了6倍以上。

解決方案二：過(guò)濾少數(shù)導(dǎo)致傾斜的key

方案適用場(chǎng)景：如果發(fā)現(xiàn)導(dǎo)致傾斜的key就少數(shù)幾個(gè)，而且對(duì)計(jì)算本身的影響并不大的話，那么很適合使用這種方案。比如99%的key就對(duì)應(yīng)10條數(shù)據(jù)，但是只有一個(gè)key對(duì)應(yīng)了100萬(wàn)數(shù)據(jù)，從而導(dǎo)致了數(shù)據(jù)傾斜。

方案實(shí)現(xiàn)思路：如果我們判斷那少數(shù)幾個(gè)數(shù)據(jù)量特別多的key，對(duì)作業(yè)的執(zhí)行和計(jì)算結(jié)果不是特別重要的話，那么干脆就直接過(guò)濾掉那少數(shù)幾個(gè)key。比如，在Spark SQL中可以使用where子句過(guò)濾掉這些key或者在Spark Core中對(duì)RDD執(zhí)行filter算子過(guò)濾掉這些key。如果需要每次作業(yè)執(zhí)行時(shí)，動(dòng)態(tài)判定哪些key的數(shù)據(jù)量最多然后再進(jìn)行過(guò)濾，那么可以使用sample算子對(duì)RDD進(jìn)行采樣，然后計(jì)算出每個(gè)key的數(shù)量，取數(shù)據(jù)量最多的key過(guò)濾掉即可。

方案實(shí)現(xiàn)原理：將導(dǎo)致數(shù)據(jù)傾斜的key給過(guò)濾掉之后，這些key就不會(huì)參與計(jì)算了，自然不可能產(chǎn)生數(shù)據(jù)傾斜。

方案優(yōu)點(diǎn)：實(shí)現(xiàn)簡(jiǎn)單，而且效果也很好，可以完全規(guī)避掉數(shù)據(jù)傾斜。

方案缺點(diǎn)：適用場(chǎng)景不多，大多數(shù)情況下，導(dǎo)致傾斜的key還是很多的，并不是只有少數(shù)幾個(gè)。

方案實(shí)踐經(jīng)驗(yàn)：在項(xiàng)目中我們也采用過(guò)這種方案解決數(shù)據(jù)傾斜。有一次發(fā)現(xiàn)某一天Spark作業(yè)在運(yùn)行的時(shí)候突然OOM了，追查之后發(fā)現(xiàn)，是Hive表中的某一個(gè)key在那天數(shù)據(jù)異常，導(dǎo)致數(shù)據(jù)量暴增。因此就采取每次執(zhí)行前先進(jìn)行采樣，計(jì)算出樣本中數(shù)據(jù)量最大的幾個(gè)key之后，直接在程序中將那些key給過(guò)濾掉。

解決方案三：提高shuffle操作的并行度

方案適用場(chǎng)景：如果我們必須要對(duì)數(shù)據(jù)傾斜迎難而上，那么建議優(yōu)先使用這種方案，因?yàn)檫@是處理數(shù)據(jù)傾斜最簡(jiǎn)單的一種方案。

方案實(shí)現(xiàn)思路：在對(duì)RDD執(zhí)行shuffle算子時(shí)，給shuffle算子傳入一個(gè)參數(shù)，比如reduceByKey(1000)，該參數(shù)就設(shè)置了這個(gè)shuffle算子執(zhí)行時(shí)shuffle read task的數(shù)量。對(duì)于Spark SQL中的shuffle類語(yǔ)句，比如group by、join等，需要設(shè)置一個(gè)參數(shù)，即spark.sql.shuffle.partitions，該參數(shù)代表了shuffle read task的并行度，該值默認(rèn)是200，對(duì)于很多場(chǎng)景來(lái)說(shuō)都有點(diǎn)過(guò)小。

方案實(shí)現(xiàn)原理：增加shuffle read task的數(shù)量，可以讓原本分配給一個(gè)task的多個(gè)key分配給多個(gè)task，從而讓每個(gè)task處理比原來(lái)更少的數(shù)據(jù)。舉例來(lái)說(shuō)，如果原本有5個(gè)key，每個(gè)key對(duì)應(yīng)10條數(shù)據(jù)，這5個(gè)key都是分配給一個(gè)task的，那么這個(gè)task就要處理50條數(shù)據(jù)。而增加了shuffle read task以后，每個(gè)task就分配到一個(gè)key，即每個(gè)task就處理10條數(shù)據(jù)，那么自然每個(gè)task的執(zhí)行時(shí)間都會(huì)變短了。具體原理如下圖所示。

方案優(yōu)點(diǎn)：實(shí)現(xiàn)起來(lái)比較簡(jiǎn)單，可以有效緩解和減輕數(shù)據(jù)傾斜的影響。

方案缺點(diǎn)：只是緩解了數(shù)據(jù)傾斜而已，沒(méi)有徹底根除問(wèn)題，根據(jù)實(shí)踐經(jīng)驗(yàn)來(lái)看，其效果有限。

方案實(shí)踐經(jīng)驗(yàn)：該方案通常無(wú)法徹底解決數(shù)據(jù)傾斜，因?yàn)槿绻霈F(xiàn)一些極端情況，比如某個(gè)key對(duì)應(yīng)的數(shù)據(jù)量有100萬(wàn)，那么無(wú)論你的task數(shù)量增加到多少，這個(gè)對(duì)應(yīng)著100萬(wàn)數(shù)據(jù)的key肯定還是會(huì)分配到一個(gè)task中去處理，因此注定還是會(huì)發(fā)生數(shù)據(jù)傾斜的。所以這種方案只能說(shuō)是在發(fā)現(xiàn)數(shù)據(jù)傾斜時(shí)嘗試使用的第一種手段，嘗試去用嘴簡(jiǎn)單的方法緩解數(shù)據(jù)傾斜而已，或者是和其他方案結(jié)合起來(lái)使用。

解決方案四：兩階段聚合（局部聚合+全局聚合）

方案適用場(chǎng)景：對(duì)RDD執(zhí)行reduceByKey等聚合類shuffle算子或者在Spark SQL中使用group by語(yǔ)句進(jìn)行分組聚合時(shí)，比較適用這種方案。

方案實(shí)現(xiàn)思路：這個(gè)方案的核心實(shí)現(xiàn)思路就是進(jìn)行兩階段聚合。第一次是局部聚合，先給每個(gè)key都打上一個(gè)隨機(jī)數(shù)，比如10以內(nèi)的隨機(jī)數(shù)，此時(shí)原先一樣的key就變成不一樣的了，比如(hello, 1) (hello, 1) (hello, 1) (hello, 1)，就會(huì)變成(1_hello, 1) (1_hello, 1) (2_hello, 1) (2_hello, 1)。接著對(duì)打上隨機(jī)數(shù)后的數(shù)據(jù)，執(zhí)行reduceByKey等聚合操作，進(jìn)行局部聚合，那么局部聚合結(jié)果，就會(huì)變成了(1_hello, 2) (2_hello, 2)。然后將各個(gè)key的前綴給去掉，就會(huì)變成(hello,2)(hello,2)，再次進(jìn)行全局聚合操作，就可以得到最終結(jié)果了，比如(hello, 4)。

方案實(shí)現(xiàn)原理：將原本相同的key通過(guò)附加隨機(jī)前綴的方式，變成多個(gè)不同的key，就可以讓原本被一個(gè)task處理的數(shù)據(jù)分散到多個(gè)task上去做局部聚合，進(jìn)而解決單個(gè)task處理數(shù)據(jù)量過(guò)多的問(wèn)題。接著去除掉隨機(jī)前綴，再次進(jìn)行全局聚合，就可以得到最終的結(jié)果。具體原理見(jiàn)下圖。

方案優(yōu)點(diǎn)：對(duì)于聚合類的shuffle操作導(dǎo)致的數(shù)據(jù)傾斜，效果是非常不錯(cuò)的。通常都可以解決掉數(shù)據(jù)傾斜，或者至少是大幅度緩解數(shù)據(jù)傾斜，將Spark作業(yè)的性能提升數(shù)倍以上。

方案缺點(diǎn)：僅僅適用于聚合類的shuffle操作，適用范圍相對(duì)較窄。如果是join類的shuffle操作，還得用其他的解決方案。

// 第一步，給RDD中的每個(gè)key都打上一個(gè)隨機(jī)前綴。
JavaPairRDD<String, Long> randomPrefixRdd = rdd.mapToPair(
        new PairFunction<Tuple2<Long,Long>, String, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<String, Long> call(Tuple2<Long, Long> tuple)
                    throws Exception {
                Random random = new Random();
                int prefix = random.nextInt(10);
                return new Tuple2<String, Long>(prefix + "_" + tuple._1, tuple._2);
            }
        });
  
// 第二步，對(duì)打上隨機(jī)前綴的key進(jìn)行局部聚合。
JavaPairRDD<String, Long> localAggrRdd = 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中每個(gè)key的隨機(jī)前綴。
JavaPairRDD<Long, Long> removedRandomPrefixRdd = localAggrRdd.mapToPair(
        new PairFunction<Tuple2<String,Long>, Long, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<Long, Long> call(Tuple2<String, Long> tuple)
                    throws Exception {
                long originalKey = Long.valueOf(tuple._1.split("_")[1]);
                return new Tuple2<Long, Long>(originalKey, tuple._2);
            }
        });
  
// 第四步，對(duì)去除了隨機(jī)前綴的RDD進(jìn)行全局聚合。
JavaPairRDD<Long, Long> globalAggrRdd = removedRandomPrefixRdd.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;
            }
        });

// 首先將數(shù)據(jù)量比較小的RDD的數(shù)據(jù)，collect到Driver中來(lái)。
List<Tuple2<Long, Row>> rdd1Data = rdd1.collect()
// 然后使用Spark的廣播功能，將小RDD的數(shù)據(jù)轉(zhuǎn)換成廣播變量，這樣每個(gè)Executor就只有一份RDD的數(shù)據(jù)。
// 可以盡可能節(jié)省內(nèi)存空間，并且減少網(wǎng)絡(luò)傳輸性能開(kāi)銷。
final Broadcast<List<Tuple2<Long, Row>>> rdd1DataBroadcast = sc.broadcast(rdd1Data);
  
// 對(duì)另外一個(gè)RDD執(zhí)行map類操作，而不再是join類操作。
JavaPairRDD<String, Tuple2<String, Row>> joinedRdd = rdd2.mapToPair(
        new PairFunction<Tuple2<Long,String>, String, Tuple2<String, Row>>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<String, Tuple2<String, Row>> call(Tuple2<Long, String> tuple)
                    throws Exception {
                // 在算子函數(shù)中，通過(guò)廣播變量，獲取到本地Executor中的rdd1數(shù)據(jù)。
                List<Tuple2<Long, Row>> rdd1Data = rdd1DataBroadcast.value();
                // 可以將rdd1的數(shù)據(jù)轉(zhuǎn)換為一個(gè)Map，便于后面進(jìn)行join操作。
                Map<Long, Row> rdd1DataMap = new HashMap<Long, Row>();
                for(Tuple2<Long, Row> data : rdd1Data) {
                    rdd1DataMap.put(data._1, data._2);
                }
                // 獲取當(dāng)前RDD數(shù)據(jù)的key以及value。
                String key = tuple._1;
                String value = tuple._2;
                // 從rdd1數(shù)據(jù)Map中，根據(jù)key獲取到可以join到的數(shù)據(jù)。
                Row rdd1Value = rdd1DataMap.get(key);
                return new Tuple2<String, String>(key, new Tuple2<String, Row>(value, rdd1Value));
            }
        });
  
// 這里得提示一下。
// 上面的做法，僅僅適用于rdd1中的key沒(méi)有重復(fù)，全部是唯一的場(chǎng)景。
// 如果rdd1中有多個(gè)相同的key，那么就得用flatMap類的操作，在進(jìn)行join的時(shí)候不能用map，而是得遍歷rdd1所有數(shù)據(jù)進(jìn)行join。
// rdd2中每條數(shù)據(jù)都可能會(huì)返回多條join后的數(shù)據(jù)。// 首先從包含了少數(shù)幾個(gè)導(dǎo)致數(shù)據(jù)傾斜key的rdd1中，采樣10%的樣本數(shù)據(jù)。
JavaPairRDD<Long, String> sampledRDD = rdd1.sample(false, 0.1);
  
// 對(duì)樣本數(shù)據(jù)RDD統(tǒng)計(jì)出每個(gè)key的出現(xiàn)次數(shù)，并按出現(xiàn)次數(shù)降序排序。
// 對(duì)降序排序后的數(shù)據(jù)，取出top 1或者top 100的數(shù)據(jù)，也就是key最多的前n個(gè)數(shù)據(jù)。
// 具體取出多少個(gè)數(shù)據(jù)量最多的key，由大家自己決定，我們這里就取1個(gè)作為示范。
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中分拆出導(dǎo)致數(shù)據(jù)傾斜的key，形成獨(dú)立的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中分拆出不導(dǎo)致數(shù)據(jù)傾斜的普通key，形成獨(dú)立的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，就是那個(gè)所有key的分布相對(duì)較為均勻的rdd。
// 這里將rdd2中，前面獲取到的key對(duì)應(yīng)的數(shù)據(jù)，過(guò)濾出來(lái)，分拆成單獨(dú)的rdd，并對(duì)rdd中的數(shù)據(jù)使用flatMap算子都擴(kuò)容100倍。
// 對(duì)擴(kuò)容的每條數(shù)據(jù)，都打上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中分拆出來(lái)的導(dǎo)致傾斜的key的獨(dú)立rdd，每條數(shù)據(jù)都打上100以內(nèi)的隨機(jī)前綴。
// 然后將這個(gè)rdd1中分拆出來(lái)的獨(dú)立rdd，與上面rdd2中分拆出來(lái)的獨(dú)立rdd，進(jìn)行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中分拆出來(lái)的包含普通key的獨(dú)立rdd，直接與rdd2進(jìn)行join。
JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD2 = commonRDD.join(rdd2);
 
// 將傾斜key join后的結(jié)果與普通key join后的結(jié)果，uinon起來(lái)。
// 就是最終的join結(jié)果。
JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD = joinedRDD1.union(joinedRDD2);// 首先從包含了少數(shù)幾個(gè)導(dǎo)致數(shù)據(jù)傾斜key的rdd1中，采樣10%的樣本數(shù)據(jù)。
JavaPairRDD<Long, String> sampledRDD = rdd1.sample(false, 0.1);
  
// 對(duì)樣本數(shù)據(jù)RDD統(tǒng)計(jì)出每個(gè)key的出現(xiàn)次數(shù)，并按出現(xiàn)次數(shù)降序排序。
// 對(duì)降序排序后的數(shù)據(jù)，取出top 1或者top 100的數(shù)據(jù)，也就是key最多的前n個(gè)數(shù)據(jù)。
// 具體取出多少個(gè)數(shù)據(jù)量最多的key，由大家自己決定，我們這里就取1個(gè)作為示范。
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中分拆出導(dǎo)致數(shù)據(jù)傾斜的key，形成獨(dú)立的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中分拆出不導(dǎo)致數(shù)據(jù)傾斜的普通key，形成獨(dú)立的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，就是那個(gè)所有key的分布相對(duì)較為均勻的rdd。
// 這里將rdd2中，前面獲取到的key對(duì)應(yīng)的數(shù)據(jù)，過(guò)濾出來(lái)，分拆成單獨(dú)的rdd，并對(duì)rdd中的數(shù)據(jù)使用flatMap算子都擴(kuò)容100倍。
// 對(duì)擴(kuò)容的每條數(shù)據(jù)，都打上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中分拆出來(lái)的導(dǎo)致傾斜的key的獨(dú)立rdd，每條數(shù)據(jù)都打上100以內(nèi)的隨機(jī)前綴。
// 然后將這個(gè)rdd1中分拆出來(lái)的獨(dú)立rdd，與上面rdd2中分拆出來(lái)的獨(dú)立rdd，進(jìn)行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中分拆出來(lái)的包含普通key的獨(dú)立rdd，直接與rdd2進(jìn)行join。
JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD2 = commonRDD.join(rdd2);
 
// 將傾斜key join后的結(jié)果與普通key join后的結(jié)果，uinon起來(lái)。
// 就是最終的join結(jié)果。
JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD = joinedRDD1.union(joinedRDD2);

// 首先將其中一個(gè)key分布相對(duì)較為均勻的RDD膨脹100倍。
JavaPairRDD<String, Row> expandedRDD = rdd1.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 {
                List<Tuple2<String, Row>> list = new ArrayList<Tuple2<String, Row>>();
                for(int i = 0; i < 100; i++) {
                    list.add(new Tuple2<String, Row>(0 + "_" + tuple._1, tuple._2));
                }
                return list;
            }
        });
  
// 其次，將另一個(gè)有數(shù)據(jù)傾斜key的RDD，每條數(shù)據(jù)都打上100以內(nèi)的隨機(jī)前綴。
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);
            }
        });
  
// 將兩個(gè)處理后的RDD進(jìn)行join即可。
JavaPairRDD<String, Tuple2<String, Row>> joinedRDD = mappedRDD.join(expandedRDD);

以下是Shffule過(guò)程中的一些主要參數(shù)，這里詳細(xì)講解了各個(gè)參數(shù)的功能、默認(rèn)值以及基于實(shí)踐經(jīng)驗(yàn)給出的調(diào)優(yōu)建議。

spark.shuffle.file.buffer

默認(rèn)值：32k

參數(shù)說(shuō)明：該參數(shù)用于設(shè)置shuffle write task的BufferedOutputStream的buffer緩沖大小。將數(shù)據(jù)寫到磁盤文件之前，會(huì)先寫入buffer緩沖中，待緩沖寫滿之后，才會(huì)溢寫到磁盤。

調(diào)優(yōu)建議：如果作業(yè)可用的內(nèi)存資源較為充足的話，可以適當(dāng)增加這個(gè)參數(shù)的大小（比如64k），從而減少shuffle write過(guò)程中溢寫磁盤文件的次數(shù)，也就可以減少磁盤IO次數(shù)，進(jìn)而提升性能。在實(shí)踐中發(fā)現(xiàn)，合理調(diào)節(jié)該參數(shù)，性能會(huì)有1%~5%的提升。

spark.reducer.maxSizeInFlight

默認(rèn)值：48m

參數(shù)說(shuō)明：該參數(shù)用于設(shè)置shuffle read task的buffer緩沖大小，而這個(gè)buffer緩沖決定了每次能夠拉取多少數(shù)據(jù)。

調(diào)優(yōu)建議：如果作業(yè)可用的內(nèi)存資源較為充足的話，可以適當(dāng)增加這個(gè)參數(shù)的大小（比如96m），從而減少拉取數(shù)據(jù)的次數(shù)，也就可以減少網(wǎng)絡(luò)傳輸?shù)拇螖?shù)，進(jìn)而提升性能。在實(shí)踐中發(fā)現(xiàn)，合理調(diào)節(jié)該參數(shù)，性能會(huì)有1%~5%的提升。

spark.shuffle.io.maxRetries

默認(rèn)值：3

參數(shù)說(shuō)明：shuffle read task從shuffle write task所在節(jié)點(diǎn)拉取屬于自己的數(shù)據(jù)時(shí)，如果因?yàn)榫W(wǎng)絡(luò)異常導(dǎo)致拉取失敗，是會(huì)自動(dòng)進(jìn)行重試的。該參數(shù)就代表了可以重試的最大次數(shù)。如果在指定次數(shù)之內(nèi)拉取還是沒(méi)有成功，就可能會(huì)導(dǎo)致作業(yè)執(zhí)行失敗。

調(diào)優(yōu)建議：對(duì)于那些包含了特別耗時(shí)的shuffle操作的作業(yè)，建議增加重試最大次數(shù)（比如60次），以避免由于JVM的full gc或者網(wǎng)絡(luò)不穩(wěn)定等因素導(dǎo)致的數(shù)據(jù)拉取失敗。在實(shí)踐中發(fā)現(xiàn)，對(duì)于針對(duì)超大數(shù)據(jù)量（數(shù)十億~上百億）的shuffle過(guò)程，調(diào)節(jié)該參數(shù)可以大幅度提升穩(wěn)定性。

spark.shuffle.io.retryWait

默認(rèn)值：5s

參數(shù)說(shuō)明：具體解釋同上，該參數(shù)代表了每次重試?yán)?shù)據(jù)的等待間隔，默認(rèn)是5s。

調(diào)優(yōu)建議：建議加大間隔時(shí)長(zhǎng)（比如60s），以增加shuffle操作的穩(wěn)定性。

spark.shuffle.memoryFraction

默認(rèn)值：0.2

參數(shù)說(shuō)明：該參數(shù)代表了Executor內(nèi)存中，分配給shuffle read task進(jìn)行聚合操作的內(nèi)存比例，默認(rèn)是20%。

調(diào)優(yōu)建議：在資源參數(shù)調(diào)優(yōu)中講解過(guò)這個(gè)參數(shù)。如果內(nèi)存充足，而且很少使用持久化操作，建議調(diào)高這個(gè)比例，給shuffle read的聚合操作更多內(nèi)存，以避免由于內(nèi)存不足導(dǎo)致聚合過(guò)程中頻繁讀寫磁盤。在實(shí)踐中發(fā)現(xiàn)，合理調(diào)節(jié)該參數(shù)可以將性能提升10%左右。

spark.shuffle.manager

默認(rèn)值：sort

參數(shù)說(shuō)明：該參數(shù)用于設(shè)置ShuffleManager的類型。Spark 1.5以后，有三個(gè)可選項(xiàng)：hash、sort和tungsten-sort。HashShuffleManager是Spark 1.2以前的默認(rèn)選項(xiàng)，但是Spark 1.2以及之后的版本默認(rèn)都是SortShuffleManager了。tungsten-sort與sort類似，但是使用了tungsten計(jì)劃中的堆外內(nèi)存管理機(jī)制，內(nèi)存使用效率更高。

調(diào)優(yōu)建議：由于SortShuffleManager默認(rèn)會(huì)對(duì)數(shù)據(jù)進(jìn)行排序，因此如果你的業(yè)務(wù)邏輯中需要該排序機(jī)制的話，則使用默認(rèn)的SortShuffleManager就可以；而如果你的業(yè)務(wù)邏輯不需要對(duì)數(shù)據(jù)進(jìn)行排序，那么建議參考后面的幾個(gè)參數(shù)調(diào)優(yōu)，通過(guò)bypass機(jī)制或優(yōu)化的HashShuffleManager來(lái)避免排序操作，同時(shí)提供較好的磁盤讀寫性能。這里要注意的是，tungsten-sort要慎用，因?yàn)橹鞍l(fā)現(xiàn)了一些相應(yīng)的bug。

spark.shuffle.sort.bypassMergeThreshold

默認(rèn)值：200

參數(shù)說(shuō)明：當(dāng)ShuffleManager為SortShuffleManager時(shí)，如果shuffle read task的數(shù)量小于這個(gè)閾值（默認(rèn)是200），則shuffle write過(guò)程中不會(huì)進(jìn)行排序操作，而是直接按照未經(jīng)優(yōu)化的HashShuffleManager的方式去寫數(shù)據(jù)，但是最后會(huì)將每個(gè)task產(chǎn)生的所有臨時(shí)磁盤文件都合并成一個(gè)文件，并會(huì)創(chuàng)建單獨(dú)的索引文件。

調(diào)優(yōu)建議：當(dāng)你使用SortShuffleManager時(shí)，如果的確不需要排序操作，那么建議將這個(gè)參數(shù)調(diào)大一些，大于shuffle read task的數(shù)量。那么此時(shí)就會(huì)自動(dòng)啟用bypass機(jī)制，map-side就不會(huì)進(jìn)行排序了，減少了排序的性能開(kāi)銷。但是這種方式下，依然會(huì)產(chǎn)生大量的磁盤文件，因此shuffle write性能有待提高。

spark.shuffle.consolidateFiles

默認(rèn)值：false

參數(shù)說(shuō)明：如果使用HashShuffleManager，該參數(shù)有效。如果設(shè)置為true，那么就會(huì)開(kāi)啟consolidate機(jī)制，會(huì)大幅度合并shuffle write的輸出文件，對(duì)于shuffle read task數(shù)量特別多的情況下，這種方法可以極大地減少磁盤IO開(kāi)銷，提升性能。

調(diào)優(yōu)建議：如果的確不需要SortShuffleManager的排序機(jī)制，那么除了使用bypass機(jī)制，還可以嘗試將spark.shffle.manager參數(shù)手動(dòng)指定為hash，使用HashShuffleManager，同時(shí)開(kāi)啟consolidate機(jī)制。在實(shí)踐中嘗試過(guò)，發(fā)現(xiàn)其性能比開(kāi)啟了bypass機(jī)制的SortShuffleManager要高出10%~30%。

本文分別講解了開(kāi)發(fā)過(guò)程中的優(yōu)化原則、運(yùn)行前的資源參數(shù)設(shè)置調(diào)優(yōu)、運(yùn)行中的數(shù)據(jù)傾斜的解決方案、為了精益求精的shuffle調(diào)優(yōu)。希望大家能夠在閱讀本文之后，記住這些性能調(diào)優(yōu)的原則以及方案，在Spark作業(yè)開(kāi)發(fā)、測(cè)試以及運(yùn)行的過(guò)程中多嘗試，只有這樣，我們才能開(kāi)發(fā)出更優(yōu)的Spark作業(yè)，不斷提升其性能。

本文鏈接：https://tech.meituan.com/2016/05/12/spark-tuning-pro.html

猜你喜歡

數(shù)倉(cāng)建模分層理論

數(shù)倉(cāng)架構(gòu)發(fā)展史

Hive整合Hbase

Hive中的鎖的用法場(chǎng)景

數(shù)倉(cāng)建模方法論

Hive表的基本操作(必會(huì))