Spark性能優化：數據傾斜調優

時間  2019-11-14
標籤 spark 性能 優化 數據 傾斜 欄目 Spark 简体版
原文   原文鏈接

前言

   繼《Spark性能優化：開發調優篇》和《Spark性能優化：資源調優篇》講解了每一個Spark開發人員都必須熟知的開發調優與資源調優以後，本文做爲《Spark性能優化指南》的高級篇，將深刻分析數據傾斜調優與shuffle調優，以解決更加棘手的性能問題。html

1.數據傾斜調優

調優概述

      有的時候，咱們可能會遇到大數據計算中一個最棘手的問題——數據傾斜，此時Spark做業的性能會比指望差不少。數據傾斜調優，就是使用各類技術方案解決不一樣類型的數據傾斜問題，以保證Spark做業的性能。java

數據傾斜發生時的現象

  • 絕大多數task執行得都很是快，但個別task執行極慢。好比，總共有1000個task，997個task都在1分鐘以內執行完了，可是剩餘兩三個task卻要一兩個小時。這種狀況很常見。sql

  • 本來可以正常執行的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所決定。app

數據傾斜原理

如何定位致使數據傾斜的代碼

      數據傾斜只會發生在shuffle過程當中。這裏給你們羅列一些經常使用的而且可能會觸發shuffle操做的算子：distinct、groupByKey、reduceByKey、aggregateByKey、join、cogroup、repartition等。出現數據傾斜時，可能就是你的代碼中使用了這些算子中的某一個所致使的。dom

某個task執行特別慢的狀況

      首先要看的，就是數據傾斜發生在第幾個stage中。ide

      若是是用yarn-client模式提交，那麼本地是直接能夠看到log的，能夠在log中找到當前運行到了第幾個stage；若是是用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上，供咱們遍歷和打印輸出。
    [plain]  view plain  copy
     
     print?在CODE上查看代碼片派生到個人代碼片
    1. val conf = new SparkConf()  
    2. val sc = new SparkContext(conf)  
    3.   
    4. val lines = sc.textFile("hdfs://...")  
    5. val words = lines.flatMap(_.split(" "))  
    6. val pairs = words.map((_, 1))  
    7. val wordCounts = pairs.reduceByKey(_ + _)  
    8.   
    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分佈的方式：

  1. 若是是Spark SQL中的group by、join語句致使的數據傾斜，那麼就查詢一下SQL中使用的表的key分佈狀況。
  2. 若是是對Spark RDD執行shuffle算子致使的數據傾斜，那麼能夠在Spark做業中加入查看key分佈的代碼，好比RDD.countByKey()。而後對統計出來的各個key出現的次數，collect/take到客戶端打印一下，就能夠看到key的分佈狀況。

      舉例來講，對於上面所說的單詞計數程序，若是肯定了是stage1的reduceByKey算子致使了數據傾斜，那麼就應該看看進行reduceByKey操做的RDD中的key分佈狀況，在這個例子中指的就是pairs RDD。以下示例，咱們能夠先對pairs採樣10%的樣本數據，而後使用countByKey算子統計出每一個key出現的次數，最後在客戶端遍歷和打印樣本數據中各個key的出現次數。

[plain]  view plain  copy
 
 print?在CODE上查看代碼片派生到個人代碼片
  1. val sampledPairs = pairs.sample(false, 0.1)  
  2. val sampledWordCounts = sampledPairs.countByKey()  
  3. sampledWordCounts.foreach(println(_))  

 

2.數據傾斜的解決方案

解決方案一：使用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操做，還得用其餘的解決方案。

[plain]  view plain  copy
 
 print?在CODE上查看代碼片派生到個人代碼片
  1. <span style="font-size:12px;">// 第一步，給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.         });  
  13.   
  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.         });  
  23.   
  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.         });  
  35.   
  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.         });</span>  

 

解決方案五：將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以上，那麼就可能發生內存溢出了。所以並不適合兩個都是大表的狀況。

 

[plain]  view plain  copy
 
 print?在CODE上查看代碼片派生到個人代碼片
  1. <span style="font-size:12px;">// 首先將數據量比較小的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);    
  6.     
  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.         });    
  29.     
  30. // 這裏得提示一下。    
  31. // 上面的作法，僅僅適用於rdd1中的key沒有重複，所有是惟一的場景。    
  32. // 若是rdd1中有多個相同的key，那麼就得用flatMap類的操做，在進行join的時候不能用map，而是得遍歷rdd1全部數據進行join。    
  33. // rdd2中每條數據均可能會返回多條join後的數據。</span>  


 

解決方案六：採樣傾斜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都致使數據傾斜，那麼這種方式也不適合。

 

[plain]  view plain  copy
 
 print?在CODE上查看代碼片派生到個人代碼片

// 首先從包含了少數幾個致使數據傾斜key的rdd1中，採樣10%的樣本數據。    

  1. JavaPairRDD<Long, String> sampledRDD = rdd1.sample(false, 0.1);    
  2. // 對樣本數據RDD統計出每一個key的出現次數，並按出現次數降序排序。    
  3. // 對降序排序後的數據，取出top 1或者top 100的數據，也就是key最多的前n個數據。    
  4. // 具體取出多少個數據量最多的key，由你們本身決定，咱們這裏就取1個做爲示範。    
  5. JavaPairRDD<Long, Long> mappedSampledRDD = sampledRDD.mapToPair(    
  6. new PairFunction<Tuple2<Long,String>, Long, Long>() {    
  7. private static final long serialVersionUID = 1L;    
  8. @Override    
  9. public Tuple2<Long, Long> call(Tuple2<Long, String> tuple)    
  10. throws Exception {    
  11. return new Tuple2<Long, Long>(tuple._1, 1L);    
  12. }    
  13. });    
  14. JavaPairRDD<Long, Long> countedSampledRDD = mappedSampledRDD.reduceByKey(    
  15. new Function2<Long, Long, Long>() {    
  16. private static final long serialVersionUID = 1L;    
  17. @Override    
  18. public Long call(Long v1, Long v2) throws Exception {    
  19. return v1 + v2;    
  20. }    
  21. });    
  22. JavaPairRDD<Long, Long> reversedSampledRDD = countedSampledRDD.mapToPair(    
  23. new PairFunction<Tuple2<Long,Long>, Long, Long>() {    
  24. private static final long serialVersionUID = 1L;    
  25. @Override    
  26. public Tuple2<Long, Long> call(Tuple2<Long, Long> tuple)    
  27. throws Exception {    
  28. return new Tuple2<Long, Long>(tuple._2, tuple._1);    
  29. }    
  30. });    
  31. //take(1)取1個致使數據傾斜的數量最多的key做爲示範
  32. final Long skewedUserid = reversedSampledRDD.sortByKey(false).take(1).get(0)._2;    
  33. // 從rdd1中分拆出致使數據傾斜的key，造成獨立的RDD。    
  34. JavaPairRDD<Long, String> skewedRDD = rdd1.filter(    
  35. new Function<Tuple2<Long,String>, Boolean>() {    
  36. private static final long serialVersionUID = 1L;    
  37. @Override    
  38. public Boolean call(Tuple2<Long, String> tuple) throws Exception {    
  39. return tuple._1.equals(skewedUserid);    
  40. }    
  41. });    
  42. // 從rdd1中分拆出不致使數據傾斜的普通key，造成獨立的RDD。    
  43. JavaPairRDD<Long, String> commonRDD = rdd1.filter(    
  44. new Function<Tuple2<Long,String>, Boolean>() {    
  45. private static final long serialVersionUID = 1L;    
  46. @Override    
  47. public Boolean call(Tuple2<Long, String> tuple) throws Exception {    
  48. return !tuple._1.equals(skewedUserid);    
  49. }    
  50. });    
  51. // rdd2，就是那個全部key的分佈相對較爲均勻的rdd。    
  52. // 這裏將rdd2中，前面獲取到的key對應的數據，過濾出來，分拆成單獨的rdd，並對rdd中的數據使用flatMap算子都擴容100倍。    
  53. // 對擴容的每條數據，都打上0～100的前綴。    
  54. JavaPairRDD<String, Row> skewedRdd2 = rdd2.filter(    
  55. new Function<Tuple2<Long,Row>, Boolean>() {    
  56. private static final long serialVersionUID = 1L;    
  57. @Override    
  58. public Boolean call(Tuple2<Long, Row> tuple) throws Exception {    
  59. return tuple._1.equals(skewedUserid);    
  60. }    
  61. }).flatMapToPair(new PairFlatMapFunction<Tuple2<Long,Row>, String, Row>() {    
  62. private static final long serialVersionUID = 1L;    
  63. @Override    
  64. public Iterable<Tuple2<String, Row>> call(    
  65. Tuple2<Long, Row> tuple) throws Exception {    
  66. Random random = new Random();    
  67. List<Tuple2<String, Row>> list = new ArrayList<Tuple2<String, Row>>();    
  68. for(int i = 0; i < 100; i++) {    
  69. list.add(new Tuple2<String, Row>(i + "_" + tuple._1, tuple._2));    
  70. }    
  71. return list;    
  72. }    
  73. });    
  74. // 將rdd1中分拆出來的致使傾斜的key的獨立rdd，每條數據都打上100之內的隨機前綴。    
  75. // 而後將這個rdd1中分拆出來的獨立rdd，與上面rdd2中分拆出來的獨立rdd，進行join。    
  76. JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD1 = skewedRDD.mapToPair(    
  77. new PairFunction<Tuple2<Long,String>, String, String>() {    
  78. private static final long serialVersionUID = 1L;    
  79. @Override    
  80. public Tuple2<String, String> call(Tuple2<Long, String> tuple)    
  81. throws Exception {    
  82. Random random = new Random();    
  83. int prefix = random.nextInt(100);    
  84. return new Tuple2<String, String>(prefix + "_" + tuple._1, tuple._2);    
  85. }    
  86. })    
  87. .join(skewedUserid2infoRDD)    
  88. .mapToPair(new PairFunction<Tuple2<String,Tuple2<String,Row>>, Long, Tuple2<String, Row>>() {    
  89. private static final long serialVersionUID = 1L;    
  90. @Override    
  91. public Tuple2<Long, Tuple2<String, Row>> call(    
  92. Tuple2<String, Tuple2<String, Row>> tuple)    
  93. throws Exception {    
  94. long key = Long.valueOf(tuple._1.split("_")[1]);    
  95. return new Tuple2<Long, Tuple2<String, Row>>(key, tuple._2);    
  96. }    
  97. });    
  98. // 將rdd1中分拆出來的包含普通key的獨立rdd，直接與rdd2進行join。    
  99. JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD2 = commonRDD.join(rdd2);    
  100. // 將傾斜key join後的結果與普通key join後的結果，uinon起來。    
  101. // 就是最終的join結果。    
  102. 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倍。

 

[plain]  view plain  copy
 
 print?在CODE上查看代碼片派生到個人代碼片
  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>(i + "_" + tuple._1, tuple._2));  
  11.                 }  
  12.                 return list;  
  13.             }  
  14.         });  
  15.   
  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.         });  
  28.   
  29. // 將兩個處理後的RDD進行join便可。  
  30. JavaPairRDD<String, Tuple2<String, Row>> joinedRDD = mappedRDD.join(expandedRDD);  

 

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

      在實踐中發現，不少狀況下，若是隻是處理較爲簡單的數據傾斜場景，那麼使用上述方案中的某一種基本就能夠解決。可是若是要處理一個較爲複雜的數據傾斜場景，那麼可能須要將多種方案組合起來使用。好比說，咱們針對出現了多個數據傾斜環節的Spark做業，能夠先運用解決方案一和二，預處理一部分數據，並過濾一部分數據來緩解；其次能夠對某些shuffle操做提高並行度，優化其性能；最後還能夠針對不一樣的聚合或join操做，選擇一種方案來優化其性能。你們須要對這些方案的思路和原理都透徹理解以後，在實踐中根據各類不一樣的狀況，靈活運用多種方案，來解決本身的數據傾斜問題。

本文轉載自：http://tech.meituan.com/spark-tuning-basic.html 

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程