Spark調優指南

Spark相關問題

1. Spark比MR快的原因？

1) Spark的計算結果可以放入記憶體，支持基于記憶體的迭代，MR不支持，

2) Spark有DAG有向無環圖，可以實作pipeline的計算模式，

3) 資源調度模式：Spark粗粒度資源調度，MR是細粒度資源調度，

資源復用：Spark中的task可以復用同一批Executor的資源，

MR里面每一個map task對應一個jvm，不能復用資源，

1. Spark中主要行程的作用？

Driver行程：負責任務的分發和結果的回收，

Executor行程：負責具體任務的執行，

Master行程：Spark資源管理的主行程，負責資源調度，

Worker行程：Spark資源管理的從行程，woker節點主要運行Executor

2. Spark調優

1. 資源調優

1) .搭建Spark集群的時候要給Spark集群足夠的資源（core，memory）

在spark安裝包的conf下spark-env.sh

SPARK_WORKER_CORES

SPARK_WORKER_MEMORY

SPARK_WORKER_INSTANCE

2) .在提交Application的時候給Application分配更多的資源，

提交命令選項：（在提交Application的時候使用選項）

--executor-cores

--executor-memory

--total-executor-cores

配置資訊：（在Application的代碼中設定

在Spark-default.conf中設定）

spark.executor.cores

spark.executor.memory

spark.max.cores

1. 并行度調優

原則：一個core一般分配2~3個task,每一個task一般處理1G資料（task的復雜度類似wc）

提高并行度的方式：

1) .如果讀取的資料在HDFS上，降低block塊的大小

2) .sc.textFile(path,numPartitions)

3) sc.parallelize(list,numPartitions) 一般用于測驗

4) coalesce、repartition可以提高RDD的磁區數，

5) 配置資訊：

spark.default.parallelism  not set (默認executor core的總個數)

spark.sql.shuffle.partitions 200

6) 自定義磁區器

2. 代碼調優
   1. 避免創建重復的RDD，復用同一個RDD
   2. 對多次使用的RDD進行持久化

如何選擇一種最合適的持久化策略？

默認情況下，性能最高的當然是MEMORY_ONLY，但前提是你的記憶體必須足夠足夠大，可以綽綽有余地存放下整個RDD的所有資料，因為不進行序列化與反序列化操作，就避免了這部分的性能開銷；對這個RDD的后續算子操作，都是基于純記憶體中的資料的操作，不需要從磁盤檔案中讀取資料，性能也很高；而且不需要復制一份資料副本，并遠程傳送到其他節點上，但是這里必須要注意的是，在實際的生產環境中，恐怕能夠直接用這種策略的場景還是有限的，如果RDD中資料比較多時（比如幾十億），直接用這種持久化級別，會導致JVM的OOM記憶體溢位例外，

如果使用MEMORY_ONLY級別時發生了記憶體溢位，那么建議嘗試使用MEMORY_ONLY_SER級別，該級別會將RDD資料序列化后再保存在記憶體中，此時每個partition僅僅是一個位元組陣列而已，大大減少了物件數量，并降低了記憶體占用，這種級別比MEMORY_ONLY多出來的性能開銷，主要就是序列化與反序列化的開銷，但是后續算子可以基于純記憶體進行操作，因此性能總體還是比較高的，此外，可能發生的問題同上，如果RDD中的資料量過多的話，還是可能會導致OOM記憶體溢位的例外，

如果純記憶體的級別都無法使用，那么建議使用MEMORY_AND_DISK_SER策略，而不是MEMORY_AND_DISK策略，因為既然到了這一步，就說明RDD的資料量很大，記憶體無法完全放下，序列化后的資料比較少，可以節省記憶體和磁盤的空間開銷，同時該策略會優先盡量嘗試將資料快取在記憶體中，記憶體快取不下才會寫入磁盤，

通常不建議使用DISK_ONLY和后綴為_2的級別：因為完全基于磁盤檔案進行資料的讀寫，會導致性能急劇降低，有時還不如重新計算一次所有RDD，后綴為_2的級別，必須將所有資料都復制一份副本，并發送到其他節點上，資料復制以及網路傳輸會導致較大的性能開銷，除非是要求作業的高可用性，否則不建議使用，

持久化算子：

cache:

MEMORY_ONLY

persist：

MEMORY_ONLY

MEMORY_ONLY_SER

MEMORY_AND_DISK_SER

一般不要選擇帶有_2的持久化級別，

checkpoint:

① 如果一個RDD的計算時間比較長或者計算起來比較復雜，一般將這個RDD的計算結果保存到HDFS上，這樣資料會更加安全，

② 如果一個RDD的依賴關系非常長，也會使用checkpoint,會切斷依賴關系，提高容錯的效率，

1. 盡量避免使用shuffle類的算子

使用廣播變數來模擬使用join,使用情況：一個RDD比較大，一個RDD比較小，

join算子=廣播變數+filter、廣播變數+map、廣播變數+flatMap

1. 使用map-side預聚合的shuffle操作

即盡量使用有combiner的shuffle類算子，

combiner概念：

在map端，每一個map task計算完畢后進行的區域聚合，

combiner好處：

1) 降低shuffle write寫磁盤的資料量，

2) 降低shuffle read拉取資料量的大小，

3) 降低reduce端聚合的次數，

有combiner的shuffle類算子：

1) reduceByKey:這個算子在map端是有combiner的，在一些場景中可以使用reduceByKey代替groupByKey，

2) aggregateByKey（fun1,func2）

1. 盡量使用高性能的算子

使用reduceByKey替代groupByKey

使用mapPartition替代map

使用foreachPartition替代foreach

filter后使用coalesce減少磁區數

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

使用repartition和coalesce算子操作磁區，

1. 使用廣播變數

開發程序中，會遇到需要在算子函式中使用外部變數的場景（尤其是大變數，比如100M以上的大集合），那么此時就應該使用Spark的廣播(Broadcast）功能來提升性能，函式中使用到外部變數時，默認情況下，Spark會將該變數復制多個副本，通過網路傳輸到task中，此時每個task都有一個變數副本，如果變數本身比較大的話（比如100M，甚至1G），那么大量的變數副本在網路中傳輸的性能開銷，以及在各個節點的Executor中占用過多記憶體導致的頻繁GC，都會極大地影響性能，如果使用的外部變數比較大，建議使用Spark的廣播功能，對該變數進行廣播，廣播后的變數，會保證每個Executor的記憶體中，只駐留一份變數副本，而Executor中的task執行時共享該Executor中的那份變數副本，這樣的話，可以大大減少變數副本的數量，從而減少網路傳輸的性能開銷，并減少對Executor記憶體的占用開銷，降低GC的頻率，

廣播大變數發送方式：Executor一開始并沒有廣播變數，而是task運行需要用到廣播變數，會找executor的blockManager要，bloackManager找Driver里面的blockManagerMaster要，

使用廣播變數可以大大降低集群中變數的副本數，不使用廣播變數，變數的副本數和task數一致，使用廣播變數變數的副本和Executor數一致，

使用廣播變數可以大大的降低集群中變數的副本數，

不使用廣播變數：變數的副本數和task數一致，

使用廣播變數:變數的副本數與Executor數一致，

廣播變數最大可以是多大?

ExecutorMemory*60%*90%*80%    = executorMemory *0.42

1. 使用Kryo優化序列化性能

在Spark中，主要有三個地方涉及到了序列化：

1) 在算子函式中使用到外部變數時，該變數會被序列化后進行網路傳輸，

2) 將自定義的型別作為RDD的泛型型別時（比如JavaRDD<SXT>，SXT是自定義型別），所有自定義型別物件，都會進行序列化，因此這種情況下，也要求自定義的類必須實作Serializable介面，

3) 使用可序列化的持久化策略時（比如MEMORY_ONLY_SER），Spark會將RDD中的每個partition都序列化成一個大的位元組陣列，

Kryo序列化器介紹：

Spark支持使用Kryo序列化機制，Kryo序列化機制，比默認的Java序列化機制，速度要快，序列化后的資料要更小，大概是Java序列化機制的1/10，所以Kryo序列化優化以后，可以讓網路傳輸的資料變少；在集群中耗費的記憶體資源大大減少，

對于這三種出現序列化的地方，我們都可以通過使用Kryo序列化類別庫，來優化序列化和反序列化的性能，Spark默認使用的是Java的序列化機制，也就是ObjectOutputStream/ObjectInputStream API來進行序列化和反序列化，但是Spark同時支持使用Kryo序列化庫，Kryo序列化類別庫的性能比Java序列化類別庫的性能要高很多，官方介紹，Kryo序列化機制比Java序列化機制，性能高10倍左右，Spark之所以默認沒有使用Kryo作為序列化類別庫，是因為Kryo要求最好要注冊所有需要進行序列化的自定義型別，因此對于開發者來說，這種方式比較麻煩，

Spark中使用Kryo：

1. 優化資料結構

java中有三種型別比較消耗記憶體：

1) 物件，每個Java物件都有物件頭、參考等額外的資訊，因此比較占用記憶體空間，

2) 字串，每個字串內部都有一個字符陣列以及長度等額外資訊，

3) 集合型別，比如HashMap、LinkedList等，因為集合型別內部通常會使用一些內部類來封裝集合元素，比如Map.Entry，

因此Spark官方建議，在Spark編碼實作中，特別是對于算子函式中的代碼，盡量不要使用上述三種資料結構，盡量使用字串替代物件，使用原始型別（比如Int、Long）替代字串，使用陣列替代集合型別，這樣盡可能地減少記憶體占用，從而降低GC頻率，提升性能，

1. 使用高性能的庫fastutil

fasteutil介紹：

fastutil是擴展了Java標準集合框架（Map、List、Set；HashMap、ArrayList、HashSet）的類別庫，提供了特殊型別的map、set、list和queue；fastutil能夠提供更小的記憶體占用，更快的存取速度；我們使用fastutil提供的集合類，來替代自己平時使用的JDK的原生的Map、List、Set，好處在于，fastutil集合類，可以減小記憶體的占用，并且在進行集合的遍歷、根據索引（或者key）獲取元素的值和設定元素的值的時候，提供更快的存取速度，fastutil的每一種集合型別，都實作了對應的Java中的標準介面（比如fastutil的map，實作了Java的Map介面），因此可以直接放入已有系統的任何代碼中，

fastutil最新版本要求Java 7以及以上版本，

使用：

見RandomExtractCars.java類

1. 資料本地化
   1. 資料本地化的級別：

1) PROCESS_LOCAL

task要計算的資料在本行程（Executor）的記憶體中，

2) NODE_LOCAL

① task所計算的資料在本節點所在的磁盤上，

② task所計算的資料在本節點其他Executor行程的記憶體中，

3) NO_PREF

task所計算的資料在關系型資料庫中，如mysql，

4) RACK_LOCAL

task所計算的資料在同機架的不同節點的磁盤或者Executor行程的記憶體中

5) ANY

跨機架，

1. Spark資料本地化調優：

Spark中任務調度時，TaskScheduler在分發之前需要依據資料的位置來分發，最好將task分發到資料所在的節點上，如果TaskScheduler分發的task在默認3s依然無法執行的話，TaskScheduler會重新發送這個task到相同的Executor中去執行，會重試5次，如果依然無法執行，那么TaskScheduler會降低一級資料本地化的級別再次發送task，

如上圖中，會先嘗試1,PROCESS_LOCAL資料本地化級別，如果重試5次每次等待3s,會默認這個Executor計算資源滿了，那么會降低一級資料本地化級別到2，NODE_LOCAL,如果還是重試5次每次等待3s還是失敗，那么還是會降低一級資料本地化級別到3，RACK_LOCAL，這樣資料就會有網路傳輸，降低了執行效率，

1) 如何提高資料本地化的級別？

可以增加每次發送task的等待時間（默認都是3s），將3s倍數調大， 結合WEBUI來調節：

• spark.locality.wait  

• spark.locality.wait.process

• spark.locality.wait.node

• spark.locality.wait.rack

注意：等待時間不能調大很大，調整資料本地化的級別不要本末倒置，雖然每一個task的本地化級別是最高了，但整個Application的執行時間反而加長，

2) 如何查看資料本地化的級別？

通過日志或者WEBUI

1. Spark Shuffle調優
   1. SparkShuffle

spark1.x 中有 兩種型別的shuffle （hashShuffleManager  另外一個是sortShuffleManager）

到spark2.x以后  只有一種shuffle 機制  SortShuffle  管理器叫做SortShuffleManager  

1. SparkShuffle概念

reduceByKey會將上一個RDD中的每一個key對應的所有value聚合成一個value，然后生成一個新的RDD，元素型別是<key,value>對的形式，這樣每一個key對應一個聚合起來的value，

問題：聚合之前，每一個key對應的value不一定都是在一個partition中，也不太可能在同一個節點上，因為RDD是分布式的彈性的資料集，RDD的partition極有可能分布在各個節點上，

如何聚合？

– Shuffle Write：上一個stage的每個map task就必須保證將自己處理的當前磁區的資料相同的key寫入一個磁區檔案中，可能會寫入多個不同的磁區檔案中，

 – Shuffle Read：reduce task就會從上一個stage的所有task所在的機器上尋找屬于己的那些磁區檔案，這樣就可以保證每一個key所對應的value都會匯聚到同一個節點上去處理和聚合，

Spark中有兩種Shuffle管理型別，HashShufflManager和SortShuffleManager，Spark1.2之前是HashShuffleManager， Spark1.2引入SortShuffleManager,在Spark 2.0+版本中已經將HashShuffleManager丟棄，

1. HashShuffleManager

1) 普通機制

• 普通機制示意圖

• 執行流程

a) 每一個map task將不同結果寫到不同的buffer中，每個buffer的大小為32K，buffer起到資料快取的作用，

b) 每個buffer檔案最后對應一個磁盤小檔案，

c) reduce task來拉取對應的磁盤小檔案，

• 總結

① .map task的計算結果會根據磁區器（默認是hashPartitioner）來決定寫入到哪一個磁盤小檔案中去，ReduceTask會去Map端拉取相應的磁盤小檔案，

② .產生的磁盤小檔案的個數：

M（map task的個數）*R（reduce task的個數）

• 存在的問題

產生的磁盤小檔案過多，會導致以下問題：

a) 在Shuffle Write程序中會產生很多寫磁盤小檔案的物件，

b) 在Shuffle Read程序中會產生很多讀取磁盤小檔案的物件，

c) 在JVM堆記憶體中物件過多會造成頻繁的gc,gc還無法解決運行所需要的記憶體 的話，就會OOM，

d) 在資料傳輸程序中會有頻繁的網路通信，頻繁的網路通信出現通信故障的可能性大大增加，一旦網路通信出現了故障會導致shuffle file cannot find 由于這個錯誤導致的task失敗，TaskScheduler不負責重試，由DAGScheduler負責重試Stage，

2) 合并機制(considation機制)

• 合并機制示意圖

• 總結

產生磁盤小檔案的個數：C(core的個數)*R（reduce的個數）

如果核數比較多的話  那么產生的小檔案個數 是不是也很多啊？

1. SortShuffleManager

1) 普通機制

• 普通機制示意圖

• 執行流程

a) map task 的計算結果會寫入到一個記憶體資料結構里面，記憶體資料結構默認是5M

b) 在shuffle的時候會有一個定時器，不定期的去估算這個記憶體結構的大小，當記憶體結構中的資料超過5M時，比如現在記憶體結構中的資料為5.01M，那么他會申請5.01*2-5=5.02M記憶體給記憶體資料結構，

c) 如果申請成功不會進行溢寫，如果申請不成功，這時候會發生溢寫磁盤，

d) 在溢寫之前記憶體結構中的資料會進行排序磁區

e) 然后開始溢寫磁盤，寫磁盤是以batch的形式去寫，一個batch是1萬條資料，

f) map task執行完成后，會將這些磁盤小檔案合并成一個大的磁盤檔案，同時生成一個索引檔案，

g) reduce task去map端拉取資料的時候，首先決議索引檔案，根據索引檔案再去拉取對應的資料，

• 總結

產生磁盤小檔案的個數： 2*M（map task的個數）

2) bypass機制

• bypass機制示意圖

• 總結

① .bypass運行機制的觸發條件如下：

shuffle reduce task的數量小于spark.shuffle.sort.bypassMergeThreshold的引數值，這個值默認是200，

② .產生的磁盤小檔案為：2*M（map task的個數）

1. Shuffle檔案尋址

1) MapOutputTracker

MapOutputTracker是Spark架構中的一個模塊，是一個主從架構，管理磁盤小檔案的地址，

• MapOutputTrackerMaster是主物件，存在于Driver中，
• MapOutputTrackerWorker是從物件，存在于Excutor中，

2) BlockManager

BlockManager塊管理者，是Spark架構中的一個模塊，也是一個主從架構，

• BlockManagerMaster,主物件，存在于Driver中，

BlockManagerMaster會在集群中有用到廣播變數和快取資料或者洗掉快取資料的時候，通知BlockManagerSlave傳輸或者洗掉資料，

• BlockManagerSlave，從物件，存在于Excutor中，

BlockManagerSlave會與BlockManagerSlave之間通信，

¬ 無論在Driver端的BlockManager還是在Excutor端的BlockManager都含有三個物件：

① DiskStore:負責磁盤的管理，

② MemoryStore：負責記憶體的管理，

③ BlockTransferService:負責資料的傳輸，

3) Shuffle檔案尋址圖

4) Shuffle檔案尋址流程

a) 當map task執行完成后，會將task的執行情況和磁盤小檔案的地址封裝到MpStatus物件中，通過MapOutputTrackerWorker物件向Driver中的MapOutputTrackerMaster匯報，

b) 在所有的map task執行完畢后，Driver中就掌握了所有的磁盤小檔案的地址，

c) 在reduce task執行之前，會通過Excutor中MapOutPutTrackerWorker向Driver端的MapOutputTrackerMaster獲取磁盤小檔案的地址，

d) 獲取到磁盤小檔案的地址后，會通過BlockManager連接資料所在節點，然后通過BlockTransferService進行資料的傳輸，

e) BlockTransferService默認啟動5個task去節點拉取資料，默認情況下，5個task拉取資料量不能超過48M，

1. Shuffle調優
   1. SparkShuffle調優配置項如何使用？

1) 在代碼中,不推薦使用，硬編碼，

new SparkConf().set(“spark.shuffle.file.buffer”,”64”)

2) 在提交spark任務的時候，推薦使用，

spark-submit --conf spark.shuffle.file.buffer=64 –conf ….

3) 在conf下的spark-default.conf組態檔中,不推薦，因為是寫死后所有應用程式都要用，

1. buffer大小
2. shuffle read拉取資料量的大小
3. shuffle聚合記憶體的比例
4. 拉取資料重試次數
5. 重試間隔時間60s
6. Spark Shuffle的種類
7. HashShuffle 合并機制
8. SortShuffle bypass機制 200次

1. Spark記憶體管理

Spark執行應用程式時，Spark集群會啟動Driver和Executor兩種JVM行程，Driver負責創建SparkContext背景關系，提交任務，task的分發等，Executor負責task的計算任務，并將結果回傳給Driver，同時需要為需要持久化的RDD提供儲存，Driver端的記憶體管理比較簡單，這里所說的Spark記憶體管理針對Executor端的記憶體管理，

Spark記憶體管理分為靜態記憶體管理和統一記憶體管理，Spark1.6之前使用的是靜態記憶體管理，Spark1.6之后引入了統一記憶體管理，

靜態記憶體管理中存盤記憶體、執行記憶體和其他記憶體的大小在 Spark 應用程式運行期間均為固定的，但用戶可以應用程式啟動前進行配置，

統一記憶體管理與靜態記憶體管理的區別在于儲存記憶體和執行記憶體共享同一塊空間，可以互相借用對方的空間，

Spark1.6以上版本默認使用的是統一記憶體管理，可以通過引數spark.memory.useLegacyMode 設定為true(默認為false)使用靜態記憶體管理，

1. 靜態記憶體管理分布圖

1. 統一記憶體管理分布圖

1. reduce 中OOM如何處理？

1) 減少每次拉取的資料量

2) 提高shuffle聚合的記憶體比例

3) 提高Excutor的總記憶體

1. 記憶體調優

比如我們創建物件  先往伊甸園和s1 中放  滿了 發生minoGC 此時 會清空 伊甸園和s1 如果還有物件 那么就往s2中放 如果s2放的下  就放在s2中  s2也滿了  會發生小型的minoGC 將物件清空

如果還有資料 將資料+1  加到15  會放入到老年代中   

但是 老年代中的物件 都是常用的物件  比如資料庫連接池等  老年代如果滿了 會發生full GC  如果清空后 還不夠用   就會發生GC

我們上面討論的問題  討論的task 的記憶體夠不夠用   

JVM堆記憶體分為一塊較大的Eden和兩塊較小的Survivor，每次只使用Eden和其中一塊Survivor，當回收時將Eden和Survivor中還存活著的物件一次性復制到另外一塊Survivor上，最后清理掉Eden和剛才用過的Survivor，也就是說當task創建出來物件會首先往Eden和survivor1中存放，survivor2是空閑的，當Eden和survivor1區域放滿以后就會觸發minor gc小型垃圾回收，清理掉不再使用的物件，會將存活下來的物件放入survivor2中，

如果存活下來的物件大小大于survivor2的大小，那么JVM就會將多余的物件直接放入到老年代中，

如果這個時候年輕代的記憶體不是很大的話，就會經常的進行minor gc，頻繁的minor gc會導致短時間內有些存活的物件（多次垃圾回收都沒有回收掉，一直在用的又不能被釋放,這種物件每經過一次minor gc都存活下來）頻繁的倒來倒去，會導致這些短生命周期的物件（不一定長期使用）每進行一次垃圾回收就會長一歲，年齡過大，默認15歲，垃圾回識訓是沒有回識訓去就會跑到老年代里面去了，

這樣會導致在老年代中存放大量的短生命周期的物件，老年代應該存放的是數量比較少并且會長期使用的物件，比如資料庫連接池物件，這樣的話，老年代就會滿溢（full gc 因為本來老年代中的物件很少，很少進行full gc 因此采取了不太復雜但是消耗性能和時間的垃圾回收演算法），不管minor gc 還是 full gc都會導致JVM的作業執行緒停止，

總結-堆記憶體不足造成的影響：

1) 頻繁的minor gc，

2) 老年代中大量的短生命周期的物件會導致full gc，

3) gc 多了就會影響Spark的性能和運行的速度，

Spark  JVM調優主要是降低gc時間，可以修改Executor記憶體的比例引數，

RDD快取、task定義運行的算子函式，可能會創建很多物件，這樣會占用大量的堆記憶體，堆記憶體滿了之后會頻繁的GC，如果GC還不能夠滿足記憶體的需要的話就會報OOM，比如一個task在運行的時候會創建N個物件，這些物件首先要放入到JVM年輕代中，比如在存資料的時候我們使用了foreach來將資料寫入到記憶體，每條資料都會封裝到一個物件中存入資料庫中，那么有多少條資料就會在JVM中創建多少個物件，

Spark中如何記憶體調優？

Spark Executor堆記憶體中存放（以靜態記憶體管理為例）：RDD的快取資料和廣播變數（spark.storage.memoryFraction 0.6），shuffle聚合記憶體（spark.shuffle.memoryFraction 0.2）,task的運行（0.2）那么如何調優呢？

1) 提高Executor總體記憶體的大小

2) 降低儲存記憶體比例或者降低聚合記憶體比例

如何查看gc？

Spark WEBUI中job->stage->task

1. 調節Executor的堆外記憶體

Spark底層shuffle的傳輸方式是使用netty傳輸，netty在進行網路傳輸的程序會申請堆外記憶體（netty是零拷貝），所以使用了堆外記憶體，默認情況下，這個堆外記憶體上限默認是每一個executor的記憶體大小的10%；真正處理大資料的時候，這里都會出現問題，導致spark作業反復崩潰，無法運行；此時就會去調節這個引數，到至少1G（1024M），甚至說2G、4G，

executor在進行shuffle write，優先從自己本地關聯的mapOutPutWorker中獲取某份資料，如果本地block manager沒有的話，那么會通過TransferService，去遠程連接其他節點上executor的block manager去獲取，嘗試建立遠程的網路連接，并且去拉取資料，頻繁創建物件讓JVM堆記憶體滿溢，進行垃圾回收，正好碰到那個exeuctor的JVM在垃圾回收，處于垃圾回程序中，所有的作業執行緒全部停止；相當于只要一旦進行垃圾回收，spark / executor停止作業，無法提供回應，spark默認的網路連接的超時時長是60s；如果卡住60s都無法建立連接的話，那么這個task就失敗了，task失敗了就會出現shuffle file cannot find的錯誤，

那么如何調節等待的時長呢？

在./spark-submit提交任務的腳本里面添加：

--conf spark.core.connection.ack.wait.timeout=300

Executor由于記憶體不足或者堆外記憶體不足了，掛掉了，對應的Executor上面的block manager也掛掉了，找不到對應的shuffle map output檔案，Reducer端不能夠拉取資料，我們可以調節堆外記憶體的大小，如何調節？

在./spark-submit提交任務的腳本里面添加

yarn下：

--conf  spark.yarn.executor.memoryOverhead=2048 單位M

standalone下：

--conf  spark.memory.offHeap.size=2048單位M

1. 解決資料傾斜
   1. 提高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的執行時間都會變短了，

1. 雙重聚合

方案適用場景：

對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處理資料量過多的問題，接著去除掉隨機前綴，再次進行全域聚合，就可以得到最終的結果，

如果一個RDD中有一個key導致資料傾斜，同時還有其他的key，那么一般先對資料集進行抽樣，然后找出傾斜的key,再使用filter對原始的RDD進行分離為兩個RDD，一個是由傾斜的key組成的RDD1，一個是由其他的key組成的RDD2，那么對于RDD1可以使用加隨機前綴進行多磁區多task計算，對于另一個RDD2正常聚合計算，最后將結果再合并起來，

1. 將reduce join轉為map join

BroadCast+filter(或者map)

方案適用場景：

在對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操作，也就不會發生資料傾斜，

1. 采樣傾斜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結果 ，

1. 使用隨機前綴和擴容RDD進行join

方案適用場景：

如果在進行join操作時，RDD中有大量的key導致資料傾斜，那么進行分拆key也沒什么意義，此時就只能使用最后一種方案來解決問題了，

方案實作思路：

該方案的實作思路基本和“解決方案六”類似，首先查看RDD/Hive表中的資料分布情況，找到那個造成資料傾斜的RDD/Hive表，比如有多個key都對應了超過1萬條資料，然后將該RDD的每條資料都打上一個n以內的隨機前綴，同時對另外一個正常的RDD進行擴容，將每條資料都擴容成n條資料，擴容出來的每條資料都依次打上一個0~n的前綴，最后將兩個處理后的RDD進行join即可，

1. Spark故障解決（troubleshooting）
   1. shuffle file cannot find：磁盤小檔案找不到，

1) connection timeout ----shuffle file cannot find

提高建立連接的超時時間，或者降低gc，降低gc了那么spark不能堆外提供服務的時間就少了，那么超時的可能就會降低，

2) fetch data fail  ---- shuffle file cannot find

提高拉取資料的重試次數以及間隔時間，

3) OOM/executor lost ---- shuffle file cannot find

提高堆外記憶體大小，提高堆內記憶體大小，

1. reduce OOM

BlockManager拉取的資料量大，reduce task處理的資料量小

解決方法：

1) 降低每次拉取的資料量

2) 提高shuffle聚合的記憶體比例

3) 提高Executor的記憶體比例

1. 序列化問題
2. Null值問題

val rdd = rdd.map{x=>{

x+”~”;

}}

rdd.foreach{x=>{

System.out.println(x.getName())

}}

標籤：大數據

上一篇：禧云數芯大資料平臺技術白皮書

下一篇：Kafka 集群在馬蜂窩大資料平臺的優化與應用擴展