Spark性能優化指南-有解無憂

1 Spark開發調優篇

原則一：避免創建重復的RDD

原則二：盡可能復用同一個RDD

原則三：對多次使用的RDD進行持久化

原則四：盡量避免使用shuffle類算子

因此在我們的開發程序中，能避免則盡可能避免使用reduceByKey、join、distinct、repartition等會進行shuffle的算子，盡量使用map類的非shuffle算子，這樣的話，沒有shuffle操作或者僅有較少shuffle操作的Spark作業，可以大大減少性能開銷，

原則五：使用map-side預聚合的shuffle操作

原則六：使用高性能的算子

1. 使用reduceByKey/aggregateByKey替代groupByKey

詳情見“原則五：使用map-side預聚合的shuffle操作”，

2. 使用mapPartitions替代普通map

mapPartitions類的算子，一次函式呼叫會處理一個partition所有的資料，而不是一次函式呼叫處理一條，性能相對來說會高一些，但是有的時候，使用mapPartitions會出現OOM（記憶體溢位）的問題，因為單次函式呼叫就要處理掉一個partition所有的資料，如果記憶體不夠，垃圾回收時是無法回收掉太多物件的，很可能出現OOM例外，所以使用這類操作時要慎重！

3. 使用foreachPartitions替代foreach

原理類似于“使用mapPartitions替代map”，也是一次函式呼叫處理一個partition的所有資料，而不是一次函式呼叫處理一條資料，在實踐中發現，foreachPartitions類的算子，對性能的提升還是很有幫助的，比如在foreach函式中，將RDD中所有資料寫MySQL，那么如果是普通的foreach算子，就會一條資料一條資料地寫，每次函式呼叫可能就會創建一個資料庫連接，此時就勢必會頻繁地創建和銷毀資料庫連接，性能是非常低下；但是如果用foreachPartitions算子一次性處理一個partition的資料，那么對于每個partition，只要創建一個資料庫連接即可，然后執行批量插入操作，此時性能是比較高的，實踐中發現，對于1萬條左右的資料量寫MySQL，性能可以提升30%以上，

4. 使用filter之后進行coalesce操作

通常對一個RDD執行filter算子過濾掉RDD中較多資料后（比如30%以上的資料），建議使用coalesce算子，手動減少RDD的partition數量，將RDD中的資料壓縮到更少的partition中去，因為filter之后，RDD的每個partition中都會有很多資料被過濾掉，此時如果照常進行后續的計算，其實每個task處理的partition中的資料量并不是很多，有一點資源浪費，而且此時處理的task越多，可能速度反而越慢，因此用coalesce減少partition數量，將RDD中的資料壓縮到更少的partition之后，只要使用更少的task即可處理完所有的partition，在某些場景下，對于性能的提升會有一定的幫助，

5. 使用repartitionAndSortWithinPartitions替代repartition與sort類操作

repartitionAndSortWithinPartitions是Spark官網推薦的一個算子，官方建議，如果需要在repartition重磁區之后，還要進行排序，建議直接使用repartitionAndSortWithinPartitions算子，因為該算子可以一邊進行重磁區的shuffle操作，一邊進行排序，shuffle與sort兩個操作同時進行，比先shuffle再sort來說，性能可能是要高的，

原則七：廣播大變數

原則八：使用Kryo優化序列化性能

原則九：優化資料結構

2 Spark資源調優篇

num-executors

引數說明：該引數用于設定Spark作業總共要用多少個Executor行程來執行，Driver在向YARN集群管理器申請資源時，YARN集群管理器會盡可能按照你的設定來在集群的各個作業節點上，啟動相應數量的Executor行程，這個引數非常之重要，如果不設定的話，默認只會給你啟動少量的Executor行程，此時你的Spark作業的運行速度是非常慢的，
引數調優建議：每個Spark作業的運行一般設定50~100個左右的Executor行程比較合適，設定太少或太多的Executor行程都不好，設定的太少，無法充分利用集群資源；設定的太多的話，大部分佇列可能無法給予充分的資源，

executor-memory

引數說明：該引數用于設定每個Executor行程的記憶體，Executor記憶體的大小，很多時候直接決定了Spark作業的性能，而且跟常見的JVM OOM例外，也有直接的關聯，
引數調優建議：每個Executor行程的記憶體設定4G_{8G較為合適，但是這只是一個參考值，具體的設定還是得根據不同部門的資源佇列來定，可以看看自己團隊的資源佇列的最大記憶體限制是多少，num-executors乘以executor-memory，是不能超過佇列的最大記憶體量的，此外，如果你是跟團隊里其他人共享這個資源佇列，那么申請的記憶體量最好不要超過資源佇列最大總記憶體的1/3}1/2，避免你自己的Spark作業占用了佇列所有的資源，導致別的同學的作業無法運行，

executor-cores

引數說明：該引數用于設定每個Executor行程的CPU core數量，這個引數決定了每個Executor行程并行執行task執行緒的能力，因為每個CPU core同一時間只能執行一個task執行緒，因此每個Executor行程的CPU core數量越多，越能夠快速地執行完分配給自己的所有task執行緒，
引數調優建議：Executor的CPU core數量設定為2~4個較為合適，同樣得根據不同部門的資源佇列來定，可以看看自己的資源佇列的最大CPU core限制是多少，再依據設定的Executor數量，來決定每個Executor行程可以分配到幾個CPU core，同樣建議，如果是跟他人共享這個佇列，那么num-executors * executor-cores不要超過佇列總CPU core的1/3~1/2左右比較合適，也是避免影響其他同學的作業運行，

driver-memory

引數說明：該引數用于設定Driver行程的記憶體，
引數調優建議：Driver的記憶體通常來說不設定，或者設定1G左右應該就夠了，唯一需要注意的一點是，如果需要使用collect算子將RDD的資料全部拉取到Driver上進行處理，那么必須確保Driver的記憶體足夠大，否則會出現OOM記憶體溢位的問題，

spark.default.parallelism

引數說明：該引數用于設定每個stage的默認task數量，這個引數極為重要，如果不設定可能會直接影響你的Spark作業性能，
引數調優建議：Spark作業的默認task數量為500~1000個較為合適，很多同學常犯的一個錯誤就是不去設定這個引數，那么此時就會導致Spark自己根據底層HDFS的block數量來設定task的數量，默認是一個HDFS block對應一個task，通常來說，Spark默認設定的數量是偏少的（比如就幾十個task），如果task數量偏少的話，就會導致你前面設定好的Executor的引數都前功盡棄，試想一下，無論你的Executor行程有多少個，記憶體和CPU有多大，但是task只有1個或者10個，那么90%的Executor行程可能根本就沒有task執行，也就是白白浪費了資源！因此Spark官網建議的設定原則是，設定該引數為num-executors * executor-cores的2~3倍較為合適，比如Executor的總CPU core數量為300個，那么設定1000個task是可以的，此時可以充分地利用Spark集群的資源，

spark.storage.memoryFraction

引數說明：該引數用于設定RDD持久化資料在Executor記憶體中能占的比例，默認是0.6，也就是說，默認Executor 60%的記憶體，可以用來保存持久化的RDD資料，根據你選擇的不同的持久化策略，如果記憶體不夠時，可能資料就不會持久化，或者資料會寫入磁盤，
引數調優建議：如果Spark作業中，有較多的RDD持久化操作，該引數的值可以適當提高一些，保證持久化的資料能夠容納在記憶體中，避免記憶體不夠快取所有的資料，導致資料只能寫入磁盤中，降低了性能，但是如果Spark作業中的shuffle類操作比較多，而持久化操作比較少，那么這個引數的值適當降低一些比較合適，此外，如果發現作業由于頻繁的gc導致運行緩慢（通過spark web ui可以觀察到作業的gc耗時），意味著task執行用戶代碼的記憶體不夠用，那么同樣建議調低這個引數的值，

spark.shuffle.memoryFraction

引數說明：該引數用于設定shuffle程序中一個task拉取到上個stage的task的輸出后，進行聚合操作時能夠使用的Executor記憶體的比例，默認是0.2，也就是說，Executor默認只有20%的記憶體用來進行該操作，shuffle操作在進行聚合時，如果發現使用的記憶體超出了這個20%的限制，那么多余的資料就會溢寫到磁盤檔案中去，此時就會極大地降低性能，
引數調優建議：如果Spark作業中的RDD持久化操作較少，shuffle操作較多時，建議降低持久化操作的記憶體占比，提高shuffle操作的記憶體占比比例，避免shuffle程序中資料過多時記憶體不夠用，必須溢寫到磁盤上，降低了性能，此外，如果發現作業由于頻繁的gc導致運行緩慢，意味著task執行用戶代碼的記憶體不夠用，那么同樣建議調低這個引數的值，

資源引數的調優，沒有一個固定的值，需要同學們根據自己的實際情況（包括Spark作業中的shuffle運算元量、RDD持久化運算元量以及spark web ui中顯示的作業gc情況），同時參考本篇文章中給出的原理以及調優建議，合理地設定上述引數，

資源引數參考示例

./bin/spark-submit 
  --master yarn-cluster 
  --num-executors 100 
  --executor-memory 6G 
  --executor-cores 4 
  --driver-memory 1G 
  --conf spark.default.parallelism=1000 
  --conf spark.storage.memoryFraction=0.5 
  --conf spark.shuffle.memoryFraction=0.3

3 Spark資料傾斜調優篇

解決方案一：使用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中去處理，因此注定還是會發生資料傾斜的，所以這種方案只能說是在發現資料傾斜時嘗試使用的第一種手段，嘗試去用嘴簡單的方法緩解資料傾斜而已，或者是和其他方案結合起來使用，

解決方案四：兩階段聚合（區域聚合+全域聚合）

方案適用場景：對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操作，還得用其他的解決方案，

// 第一步，給RDD中的每個key都打上一個隨機前綴，

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);

      }

    });

// 第二步，對打上隨機前綴的key進行區域聚合，

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中每個key的隨機前綴，

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);

      }

    });

// 第四步，對去除了隨機前綴的RDD進行全域聚合，

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;

      }

    });

解決方案五：將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以上，那么就可能發生記憶體溢位了，因此并不適合兩個都是大表的情況，

// 首先將資料量比較小的RDD的資料，collect到Driver中來，*

List<Tuple2<Long, Row>> rdd1Data = https://www.cnblogs.com/weijiqian/p/rdd1.collect()

// 然后使用Spark的廣播功能，將小RDD的資料轉換成廣播變數，這樣每個Executor就只有一份RDD的資料，

// 可以盡可能節省記憶體空間，并且減少網路傳輸性能開銷，

final Broadcast>> rdd1DataBroadcast = sc.broadcast(rdd1Data);

// 對另外一個RDD執行map類操作，而不再是join類操作，

JavaPairRDD> joinedRdd = rdd2.mapToPair(

    new PairFunction, String, Tuple2>() {

      private static final long serialVersionUID = 1L;

      @Override

      public Tuple2> call(Tuple2 tuple)

          throws Exception {

        // 在算子函式中，通過廣播變數，獲取到本地Executor中的rdd1資料，

        List> rdd1Data = rdd1DataBroadcast.value();

        // 可以將rdd1的資料轉換為一個Map，便于后面進行join操作，

        Map rdd1DataMap = new HashMap();

        for(Tuple2 data : rdd1Data) {

          rdd1DataMap.put(data.1, data.2);

        }

        // 獲取當前RDD資料的key以及value，

        String key = tuple._1;

        String value = tuple._2;

        // 從rdd1資料Map中，根據key獲取到可以join到的資料，

        Row rdd1Value = rdd1DataMap.get(key);

        return new Tuple2(key, new Tuple2(value, rdd1Value));

      }

    });

// 這里得提示一下，

// 上面的做法，僅僅適用于rdd1中的key沒有重復，全部是唯一的場景，

// 如果rdd1中有多個相同的key，那么就得用flatMap類的操作，在進行join的時候不能用map，而是得遍歷rdd1所有資料進行join，

// 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都導致資料傾斜，那么這種方式也不適合，


// 首先從包含了少數幾個導致資料傾斜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進行擴容，對記憶體資源要求很高，

方案實踐經驗：曾經開發一個資料需求的時候，發現一個join導致了資料傾斜，優化之前，作業的執行時間大約是60分鐘左右；使用該方案優化之后，執行時間縮短到10分鐘左右，性能提升了6倍，

// 首先將其中一個key分布相對較為均勻的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;

?      }

?    });

*// 其次，將另一個有資料傾斜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);

?      }

?    });

*// 將兩個處理后的RDD進行join即可，*

JavaPairRDD<String, Tuple2<String, Row>> joinedRDD = mappedRDD.join(expandedRDD);

解決方案八：多種方案組合使用

在實踐中發現，很多情況下，如果只是處理較為簡單的資料傾斜場景，那么使用上述方案中的某一種基本就可以解決，但是如果要處理一個較為復雜的資料傾斜場景，那么可能需要將多種方案組合起來使用，比如說，我們針對出現了多個資料傾斜環節的Spark作業，可以先運用解決方案一和二，預處理一部分資料，并過濾一部分資料來緩解；其次可以對某些shuffle操作提升并行度，優化其性能；最后還可以針對不同的聚合或join操作，選擇一種方案來優化其性能，大家需要對這些方案的思路和原理都透徹理解之后，在實踐中根據各種不同的情況，靈活運用多種方案，來解決自己的資料傾斜問題，

4 Spark shuffle調優篇

大多數Spark作業的性能主要就是消耗在了shuffle環節，因為該環節包含了大量的磁盤IO、序列化、網路資料傳輸等操作，因此，如果要讓作業的性能更上一層樓，就有必要對shuffle程序進行調優，但是也必須提醒大家的是，影響一個Spark作業性能的因素，主要還是代碼開發、資源引數以及資料傾斜，shuffle調優只能在整個Spark的性能調優中占到一小部分而已，因此大家務必把握住調優的基本原則，千萬不要舍本逐末，下面我們就給大家詳細講解shuffle的原理，以及相關引數的說明，同時給出各個引數的調優建議，

一、ShuffleManager發展概述

在Spark的原始碼中，負責shuffle程序的執行、計算和處理的組件主要就是ShuffleManager，也即shuffle管理器，而隨著Spark的版本的發展，ShuffleManager也在不斷迭代，變得越來越先進，

在Spark 1.2以前，默認的shuffle計算引擎是HashShuffleManager，該ShuffleManager而HashShuffleManager有著一個非常嚴重的弊端，就是會產生大量的中間磁盤檔案，進而由大量的磁盤IO操作影響了性能，

因此在Spark 1.2以后的版本中，默認的ShuffleManager改成了SortShuffleManager，SortShuffleManager相較于HashShuffleManager來說，有了一定的改進，主要就在于，每個Task在進行shuffle操作時，雖然也會產生較多的臨時磁盤檔案，但是最后會將所有的臨時檔案合并（merge）成一個磁盤檔案，因此每個Task就只有一個磁盤檔案，在下一個stage的shuffle read task拉取自己的資料時，只要根據索引讀取每個磁盤檔案中的部分資料即可，

下面我們詳細分析一下HashShuffleManager和SortShuffleManager的原理，

二、HashShuffleManager運行原理

1、未經優化的HashShuffleManager

下圖說明了未經優化的HashShuffleManager的原理，這里我們先明確一個假設前提：每個Executor只有1個CPU core，也就是說，無論這個Executor上分配多少個task執行緒，同一時間都只能執行一個task執行緒，

我們先從shuffle write開始說起，shuffle write階段，主要就是在一個stage結束計算之后，為了下一個stage可以執行shuffle類的算子（比如reduceByKey），而將每個task處理的資料按key進行“分類”，所謂“分類”，就是對相同的key執行hash演算法，從而將相同key都寫入同一個磁盤檔案中，而每一個磁盤檔案都只屬于下游stage的一個task，在將資料寫入磁盤之前，會先將資料寫入記憶體緩沖中，當記憶體緩沖填滿之后，才會溢寫到磁盤檔案中去，

那么每個執行shuffle write的task，要為下一個stage創建多少個磁盤檔案呢？很簡單，下一個stage的task有多少個，當前stage的每個task就要創建多少份磁盤檔案，比如下一個stage總共有100個task，那么當前stage的每個task都要創建100份磁盤檔案，如果當前stage有50個task，總共有10個Executor，每個Executor執行5個Task，那么每個Executor上總共就要創建500個磁盤檔案，所有Executor上會創建5000個磁盤檔案，由此可見，未經優化的shuffle write操作所產生的磁盤檔案的數量是極其驚人的，

接著我們來說說shuffle read，shuffle read，通常就是一個stage剛開始時要做的事情，此時該stage的每一個task就需要將上一個stage的計算結果中的所有相同key，從各個節點上通過網路都拉取到自己所在的節點上，然后進行key的聚合或連接等操作，由于shuffle write的程序中，task給下游stage的每個task都創建了一個磁盤檔案，因此shuffle read的程序中，每個task只要從上游stage的所有task所在節點上，拉取屬于自己的那一個磁盤檔案即可，

shuffle read的拉取程序是一邊拉取一邊進行聚合的，每個shuffle read task都會有一個自己的buffer緩沖，每次都只能拉取與buffer緩沖相同大小的資料，然后通過記憶體中的一個Map進行聚合等操作，聚合完一批資料后，再拉取下一批資料，并放到buffer緩沖中進行聚合操作，以此類推，直到最后將所有資料到拉取完，并得到最終的結果，

2、優化后的HashShuffleManager

下圖說明了優化后的HashShuffleManager的原理，這里說的優化，是指我們可以設定一個引數，spark.shuffle.consolidateFiles，該引數默認值為false，將其設定為true即可開啟優化機制，通常來說，如果我們使用HashShuffleManager，那么都建議開啟這個選項，

開啟consolidate機制之后，在shuffle write程序中，task就不是為下游stage的每個task創建一個磁盤檔案了，此時會出現shuffleFileGroup的概念，每個shuffleFileGroup會對應一批磁盤檔案，磁盤檔案的數量與下游stage的task數量是相同的，一個Executor上有多少個CPU core，就可以并行執行多少個task，而第一批并行執行的每個task都會創建一個shuffleFileGroup，并將資料寫入對應的磁盤檔案內，

當Executor的CPU core執行完一批task，接著執行下一批task時，下一批task就會復用之前已有的shuffleFileGroup，包括其中的磁盤檔案，也就是說，此時task會將資料寫入已有的磁盤檔案中，而不會寫入新的磁盤檔案中，因此，consolidate機制允許不同的task復用同一批磁盤檔案，這樣就可以有效將多個task的磁盤檔案進行一定程度上的合并，從而大幅度減少磁盤檔案的數量，進而提升shuffle write的性能，

假設第二個stage有100個task，第一個stage有50個task，總共還是有10個Executor，每個Executor執行5個task，那么原本使用未經優化的HashShuffleManager時，每個Executor會產生500個磁盤檔案，所有Executor會產生5000個磁盤檔案的，但是此時經過優化之后，每個Executor創建的磁盤檔案的數量的計算公式為：CPU core的數量 * 下一個stage的task數量，也就是說，每個Executor此時只會創建100個磁盤檔案，所有Executor只會創建1000個磁盤檔案，

三、SortShuffleManager運行原理

SortShuffleManager的運行機制主要分成兩種，一種是普通運行機制，另一種是bypass運行機制，當shuffle read task的數量小于等于spark.shuffle.sort.bypassMergeThreshold引數的值時（默認為200），就會啟用bypass機制，

四、shuffle相關引數調優

以下是Shffule程序中的一些主要引數，這里詳細講解了各個引數的功能、默認值以及基于實踐經驗給出的調優建議，

spark.shuffle.file.buffer

默認值：32k
引數說明：該引數用于設定shuffle write task的BufferedOutputStream的buffer緩沖大小，將資料寫到磁盤檔案之前，會先寫入buffer緩沖中，待緩沖寫滿之后，才會溢寫到磁盤，
調優建議：如果作業可用的記憶體資源較為充足的話，可以適當增加這個引數的大小（比如64k），從而減少shuffle write程序中溢寫磁盤檔案的次數，也就可以減少磁盤IO次數，進而提升性能，在實踐中發現，合理調節該引數，性能會有1%~5%的提升，

spark.reducer.maxSizeInFlight

默認值：48m
引數說明：該引數用于設定shuffle read task的buffer緩沖大小，而這個buffer緩沖決定了每次能夠拉取多少資料，
調優建議：如果作業可用的記憶體資源較為充足的話，可以適當增加這個引數的大小（比如96m），從而減少拉取資料的次數，也就可以減少網路傳輸的次數，進而提升性能，在實踐中發現，合理調節該引數，性能會有1%~5%的提升，

spark.shuffle.io.maxRetries

默認值：3
引數說明：shuffle read task從shuffle write task所在節點拉取屬于自己的資料時，如果因為網路例外導致拉取失敗，是會自動進行重試的，該引數就代表了可以重試的最大次數，如果在指定次數之內拉取還是沒有成功，就可能會導致作業執行失敗，
調優建議：對于那些包含了特別耗時的shuffle操作的作業，建議增加重試最大次數（比如60次），以避免由于JVM的full gc或者網路不穩定等因素導致的資料拉取失敗，在實踐中發現，對于針對超大資料量（數十億~上百億）的shuffle程序，調節該引數可以大幅度提升穩定性，

spark.shuffle.io.retryWait

默認值：5s
引數說明：具體解釋同上，該引數代表了每次重試拉取資料的等待間隔，默認是5s，
調優建議：建議加大間隔時長（比如60s），以增加shuffle操作的穩定性，

spark.shuffle.memoryFraction

默認值：0.2
引數說明：該引數代表了Executor記憶體中，分配給shuffle read task進行聚合操作的記憶體比例，默認是20%，
調優建議：在資源引數調優中講解過這個引數，如果記憶體充足，而且很少使用持久化操作，建議調高這個比例，給shuffle read的聚合操作更多記憶體，以避免由于記憶體不足導致聚合程序中頻繁讀寫磁盤，在實踐中發現，合理調節該引數可以將性能提升10%左右，

spark.shuffle.manager

默認值：sort
引數說明：該引數用于設定ShuffleManager的型別，Spark 1.5以后，有三個可選項：hash、sort和tungsten-sort，HashShuffleManager是Spark 1.2以前的默認選項，但是Spark 1.2以及之后的版本默認都是SortShuffleManager了，tungsten-sort與sort類似，但是使用了tungsten計劃中的堆外記憶體管理機制，記憶體使用效率更高，
調優建議：由于SortShuffleManager默認會對資料進行排序，因此如果你的業務邏輯中需要該排序機制的話，則使用默認的SortShuffleManager就可以；而如果你的業務邏輯不需要對資料進行排序，那么建議參考后面的幾個引數調優，通過bypass機制或優化的HashShuffleManager來避免排序操作，同時提供較好的磁盤讀寫性能，這里要注意的是，tungsten-sort要慎用，因為之前發現了一些相應的bug，

spark.shuffle.sort.bypassMergeThreshold

默認值：200
引數說明：當ShuffleManager為SortShuffleManager時，如果shuffle read task的數量小于這個閾值（默認是200），則shuffle write程序中不會進行排序操作，而是直接按照未經優化的HashShuffleManager的方式去寫資料，但是最后會將每個task產生的所有臨時磁盤檔案都合并成一個檔案，并會創建單獨的索引檔案，
調優建議：當你使用SortShuffleManager時，如果的確不需要排序操作，那么建議將這個引數調大一些，大于shuffle read task的數量，那么此時就會自動啟用bypass機制，map-side就不會進行排序了，減少了排序的性能開銷，但是這種方式下，依然會產生大量的磁盤檔案，因此shuffle write性能有待提高，

spark.shuffle.consolidateFiles

默認值：false
引數說明：如果使用HashShuffleManager，該引數有效，如果設定為true，那么就會開啟consolidate機制，會大幅度合并shuffle write的輸出檔案，對于shuffle read task數量特別多的情況下，這種方法可以極大地減少磁盤IO開銷，提升性能，
調優建議：如果的確不需要SortShuffleManager的排序機制，那么除了使用bypass機制，還可以嘗試將spark.shffle.manager引數手動指定為hash，使用HashShuffleManager，同時開啟consolidate機制，在實踐中嘗試過，發現其性能比開啟了bypass機制的SortShuffleManager要高出10%~30%，

原文:公眾號:大資料架構

轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/224939.html

標籤：大數據

上一篇：SQL 陳述句中 where 條件后寫上1=1 是什么意思

下一篇：Spark引數優化