Spark調優-第一天

1. 掌握Spark開發的十大原則
2. 掌握基于Spark記憶體模型調優
3. 帶領學生熟悉Spark官網

1. 十大開發原則（60分鐘）

1.1 避免建立重複的RDD

通常來說，我們在開發一個Spark作業時，首先是基于某個資料源（比如Hive表或HDFS檔案）建立一個初始的RDD；接着對這個RDD執行某個算子操作，然後得到下一個RDD；以此類推，循環往複，直到計算出最終我們需要的結果。在這個過程中，多個RDD會通過不同的算子操作（比如map、reduce等）串起來，這個“RDD串”，就是RDD lineage，也就是“RDD的血緣關系鍊”。

我們在開發過程中要注意：對于同一份資料，隻應該建立一個RDD，不能建立多個RDD來代表同一份資料。

一些Spark初學者在剛開始開發Spark作業時，或者是有經驗的工程師在開發RDD lineage極其冗長的Spark作業時，可能會忘了自己之前對于某一份資料已經建立過一個RDD了，進而導緻對于同一份資料，建立了多個RDD。這就意味着，我們的Spark作業會進行多次重複計算來建立多個代表相同資料的RDD，進而增加了作業的性能開銷。

// 需要對名為“hello.txt”的HDFS檔案進行一次map操作，再進行一次reduce操作。也就是說，需要對一份資料執行兩次算子操作。

// 錯誤的做法：對于同一份資料執行多次算子操作時，建立多個RDD。
// 這裡執行了兩次textFile方法，針對同一個HDFS檔案，建立了兩個RDD出來，然後分别對每個RDD都執行了一個算子操作。
// 這種情況下，Spark需要從HDFS上兩次加載hello.txt檔案的内容，并建立兩個單獨的RDD；第二次加載HDFS檔案以及建立RDD的性能開銷，很明顯是白白浪費掉的。
假設你的一個段代碼有1000行
val rdd1 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt")
rdd1.map(...)
val rdd2 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt")
rdd2.reduce(...)

// 正确的用法：對于一份資料執行多次算子操作時，隻使用一個RDD。
// 這種寫法很明顯比上一種寫法要好多了，因為我們對于同一份資料隻建立了一個RDD，然後對這一個RDD執行了多次算子操作。
// 但是要注意到這裡為止優化還沒有結束，由于rdd1被執行了兩次算子操作，第二次執行reduce操作的時候，還會再次從源頭處重新計算一次rdd1的資料，是以還是會有重複計算的性能開銷。
// 要徹底解決這個問題，必須結合“原則三：對多次使用的RDD進行持久化”，才能保證一個RDD被多次使用時隻被計算一次。
val rdd1 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt")
rdd1.map(...)
rdd1.reduce(...)
           

1.2 盡可能複用同一個RDD

除了要避免在開發過程中對一份完全相同的資料建立多個RDD之外，在對不同的資料執行算子操作時還要盡可能地複用一個RDD。比如說，有一個RDD的資料格式是key-value類型的，另一個是單value類型的，這兩個RDD的value資料是完全一樣的。那麼此時我們可以隻使用key-value類型的那個RDD，因為其中已經包含了另一個的資料。對于類似這種多個RDD的資料有重疊或者包含的情況，我們應該盡量複用一個RDD，這樣可以盡可能地減少RDD的數量，進而盡可能減少算子執行的次數。

// 錯誤的做法。

// 有一個<Long, String>格式的RDD，即rdd1。
// 接着由于業務需要，對rdd1執行了一個map操作，建立了一個rdd2，而rdd2中的資料僅僅是rdd1中的value值而已，也就是說，rdd2是rdd1的子集。
JavaPairRDD<Long, String> rdd1 = ...

JavaRDD<String> rdd2 = rdd1.map(...)

// 分别對rdd1和rdd2執行了不同的算子操作。
rdd1.reduceByKey(...)
rdd2.map(...)

// 正确的做法。

// 上面這個case中，其實rdd1和rdd2的差別無非就是資料格式不同而已，rdd2的資料完全就是rdd1的子集而已，卻建立了兩個rdd，并對兩個rdd都執行了一次算子操作。
// 此時會因為對rdd1執行map算子來建立rdd2，而多執行一次算子操作，進而增加性能開銷。

// 其實在這種情況下完全可以複用同一個RDD。
// 我們可以使用rdd1，既做reduceByKey操作，也做map操作。
// 在進行第二個map操作時，隻使用每個資料的tuple._2，也就是rdd1中的value值，即可。
JavaPairRDD<Long, String> rdd1 = ...
rdd1.reduceByKey(...)
rdd1.map(tuple._2...)

// 第二種方式相較于第一種方式而言，很明顯減少了一次rdd2的計算開銷。
// 但是到這裡為止，優化還沒有結束，對rdd1我們還是執行了兩次算子操作，rdd1實際上還是會被計算兩次。
// 是以還需要配合“原則三：對多次使用的RDD進行持久化”進行使用，才能保證一個RDD被多次使用時隻被計算一次。
           

1.3 對多次使用的RDD進行持久化

當你在Spark代碼中多次對一個RDD做了算子操作後，恭喜，你已經實作Spark作業第一步的優化了，也就是盡可能複用RDD。此時就該在這個基礎之上，進行第二步優化了，也就是要保證對一個RDD執行多次算子操作時，這個RDD本身僅僅被計算一次。

Spark中對于一個RDD執行多次算子的預設原理是這樣的：每次你對一個RDD執行一個算子操作時，都會重新從源頭處計算一遍，計算出那個RDD來，然後再對這個RDD執行你的算子操作。這種方式的性能是很差的。

是以對于這種情況，我們的建議是：對多次使用的RDD進行持久化。此時Spark就會根據你的持久化政策，将RDD中的資料儲存到記憶體或者磁盤中。以後每次對這個RDD進行算子操作時，都會直接從記憶體或磁盤中提取持久化的RDD資料，然後執行算子，而不會從源頭處重新計算一遍這個RDD，再執行算子操作。

// 如果要對一個RDD進行持久化，隻要對這個RDD調用cache()和persist()即可。

// 正确的做法。
// cache()方法表示：使用非序列化的方式将RDD中的資料全部嘗試持久化到記憶體中。
// 此時再對rdd1執行兩次算子操作時，隻有在第一次執行map算子時，才會将這個rdd1從源頭處計算一次。
// 第二次執行reduce算子時，就會直接從記憶體中提取資料進行計算，不會重複計算一個rdd。
val rdd1 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt").cache()
rdd1.map(...)
rdd1.reduce(...)

// persist()方法表示：手動選擇持久化級别，并使用指定的方式進行持久化。
// 比如說，StorageLevel.MEMORY_AND_DISK_SER表示，記憶體充足時優先持久化到記憶體中，記憶體不充足時持久化到磁盤檔案中。
// 而且其中的_SER字尾表示，使用序列化的方式來儲存RDD資料，此時RDD中的每個partition都會序列化成一個大的位元組數組，然後再持久化到記憶體或磁盤中。
// 序列化的方式可以減少持久化的資料對記憶體/磁盤的占用量，進而避免記憶體被持久化資料占用過多，進而發生頻繁GC。
val rdd1 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt").persist(StorageLevel.MEMORY_AND_DISK_SER)
rdd1.map(...)
rdd1.reduce(...)
對于persist()方法而言，我們可以根據不同的業務場景選擇不同的持久化級别。
           

Spark的持久化級别

持久化級别	含義解釋
MEMORY_ONLY	使用未序列化的Java對象格式，将資料儲存在記憶體中。如果記憶體不夠存放所有的資料，則資料可能就不會進行持久化。那麼下次對這個RDD執行算子操作時，那些沒有被持久化的資料，需要從源頭處重新計算一遍。這是預設的持久化政策，使用cache()方法時，實際就是使用的這種持久化政策。
MEMORY_AND_DISK	使用未序列化的Java對象格式，優先嘗試将資料儲存在記憶體中。如果記憶體不夠存放所有的資料，會将資料寫入磁盤檔案中，下次對這個RDD執行算子時，持久化在磁盤檔案中的資料會被讀取出來使用。
MEMORY_ONLY_SER	基本含義同MEMORY_ONLY。唯一的差別是，會将RDD中的資料進行序列化，RDD的每個partition會被序列化成一個位元組數組。這種方式更加節省記憶體，進而可以避免持久化的資料占用過多記憶體導緻頻繁GC。
MEMORY_AND_DISK_SER	基本含義同MEMORY_AND_DISK。唯一的差別是，會将RDD中的資料進行序列化，RDD的每個partition會被序列化成一個位元組數組。這種方式更加節省記憶體，進而可以避免持久化的資料占用過多記憶體導緻頻繁GC。
DISK_ONLY	使用未序列化的Java對象格式，将資料全部寫入磁盤檔案中。
MEMORY_ONLY_2, MEMORY_AND_DISK_2, 等等.	對于上述任意一種持久化政策，如果加上字尾_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的級别，必須将所有資料都複制一份副本，并發送到其他節點上，資料複制以及網絡傳輸會導緻較大的性能開銷，除非是要求作業的高可用性，否則不建議使用。

1.4 盡量避免使用shuffle類算子

如果有可能的話，要盡量避免使用shuffle類算子。因為Spark作業運作過程中，最消耗性能的地方就是shuffle過程。shuffle過程，簡單來說，就是将分布在叢集中多個節點上的同一個key，拉取到同一個節點上，進行聚合或join等操作。比如reduceByKey、join等算子，都會觸發shuffle操作。

shuffle過程中，各個節點上的相同key都會先寫入本地磁盤檔案中，然後其他節點需要通過網絡傳輸拉取各個節點上的磁盤檔案中的相同key。而且相同key都拉取到同一個節點進行聚合操作時，還有可能會因為一個節點上處理的key過多，導緻記憶體不夠存放，進而溢寫到磁盤檔案中。是以在shuffle過程中，可能會發生大量的磁盤檔案讀寫的IO操作，以及資料的網絡傳輸操作。磁盤IO和網絡資料傳輸也是shuffle性能較差的主要原因。

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

//錯誤的做法：
// 傳統的join操作會導緻shuffle操作。
// 因為兩個RDD中，相同的key都需要通過網絡拉取到一個節點上，由一個task進行join操作。
//要求其中的一個rdd資料量要小一些，不能很大，不能超過5G。
//就把小的這個rdd的資料給廣播出去
val rdd3 = rdd1.join(rdd2)
    
//正确的做法：
// Broadcast+map的join操作，不會導緻shuffle操作。
// 使用Broadcast将一個資料量較小的RDD作為廣播變量。
val rdd2Data = rdd2.collect()
val rdd2DataBroadcast = sc.broadcast(rdd2Data)

// 在rdd1.map算子中，可以從rdd2DataBroadcast中，擷取rdd2的所有資料。
// 然後進行周遊，如果發現rdd2中某條資料的key與rdd1的目前資料的key是相同的，那麼就判定可以進行join。
// 此時就可以根據自己需要的方式，将rdd1目前資料與rdd2中可以連接配接的資料，拼接在一起（String或Tuple）。
val rdd3 = rdd1.map(rdd2DataBroadcast...)

// 注意，以上操作，建議僅僅在rdd2的資料量比較少（比如幾百M，或者一兩G）的情況下使用。
// 因為每個Executor的記憶體中，都會駐留一份rdd2的全量資料。
           

這兒我們暫時不給大家示範實際的例子，後面會給大家敲一個實際的案例。

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

如果因為業務需要，一定要使用shuffle操作，無法用map類的算子來替代，那麼盡量使用可以map-side預聚合的算子。

所謂的map-side預聚合，說的是在每個節點本地對相同的key進行一次聚合操作，類似于MapReduce中的本地combiner。map-side預聚合之後，每個節點本地就隻會有一條相同的key，因為多條相同的key都被聚合起來了。其他節點在拉取所有節點上的相同key時，就會大大減少需要拉取的資料數量，進而也就減少了磁盤IO以及網絡傳輸開銷。通常來說，在可能的情況下，建議使用reduceByKey或者aggregateByKey算子來替代掉groupByKey算子。因為reduceByKey和aggregateByKey算子都會使用使用者自定義的函數對每個節點本地的相同key進行預聚合。而groupByKey算子是不會進行預聚合的，全量的資料會在叢集的各個節點之間分發和傳輸，性能相對來說比較差。

比如如下兩幅圖，就是典型的例子，分别基于reduceByKey和groupByKey進行單詞計數。其中第一張圖是groupByKey的原理圖，可以看到，沒有進行任何本地聚合時，所有資料都會在叢集節點之間傳輸；第二張圖是reduceByKey的原理圖，可以看到，每個節點本地的相同key資料，都進行了預聚合，然後才傳輸到其他節點上進行全局聚合。

groupByKey原理：

reduceBykey原理：

1.6 使用高性能的算子

除了shuffle相關的算子有優化原則之外，其他的算子也都有着相應的優化原則。

使用reduceByKey/aggregateByKey替代groupByKey

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

使用mapPartitions替代普通map

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

使用foreachPartitions替代foreach

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

使用filter之後進行coalesce操作

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

使用repartitionAndSortWithinPartitions替代repartition與sort類操作

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

這個情況使用得不多，大家了解即可，有可能我講了以後，要是沒有使用過這個repartitionAndSortWithinPartitions算子的同學，其實沒聽明白我意思。但是沒事，這個不重要。

注：coalesce：減少分區數，repartition:增加分區數

1.7 廣播大變量

有時在開發過程中，會遇到需要在算子函數中使用外部變量的場景（尤其是大變量，比如100M以上的大集合），那麼此時就應該使用Spark的廣播（Broadcast）功能來提升性能。

在算子函數中使用到外部變量時，預設情況下，Spark會将該變量複制多個副本，通過網絡傳輸到task中，此時每個task都有一個變量副本。如果變量本身比較大的話（比如100M，甚至1G），那麼大量的變量副本在網絡中傳輸的性能開銷，以及在各個節點的Executor中占用過多記憶體導緻的頻繁GC，都會極大地影響性能。

是以對于上述情況，如果使用的外部變量比較大，建議使用Spark的廣播功能，對該變量進行廣播。廣播後的變量，會保證每個Executor的記憶體中，隻駐留一份變量副本，而Executor中的task執行時共享該Executor中的那份變量副本。這樣的話，可以大大減少變量副本的數量，進而減少網絡傳輸的性能開銷，并減少對Executor記憶體的占用開銷，降低GC的頻率。

// 以下代碼在算子函數中，使用了外部的變量。
// 此時沒有做任何特殊操作，每個task都會有一份list1的副本。
val list1 = ...
rdd1.map(list1...)

// 以下代碼将list1封裝成了Broadcast類型的廣播變量。
// 在算子函數中，使用廣播變量時，首先會判斷目前task所在Executor記憶體中，是否有變量副本。
// 如果有則直接使用；如果沒有則從Driver或者其他Executor節點上遠端拉取一份放到本地Executor記憶體中。
// 每個Executor記憶體中，就隻會駐留一份廣播變量副本。
val list1 = ...
val list1Broadcast = sc.broadcast(list1)
rdd1.map(list1Broadcast...) 
           

1.8 使用Kryo優化序列化性能

spark裡面有一種預設，那種性能不好，一是序列化的時候速度慢，二 序列化後體積變化不大。

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

• 在算子函數中使用到外部變量時，該變量會被序列化後進行網絡傳輸（見“原則七：廣播大變量”中的講解）。
• 将自定義的類型作為RDD的泛型類型時（比如JavaRDD，Student是自定義類型），所有自定義類型對象，都會進行序列化。是以這種情況下，也要求自定義的類必須實作Serializable接口。
• 使用可序列化的持久化政策時（比如MEMORY_ONLY_SER），Spark會将RDD中的每個partition都序列化成一個大的位元組數組。

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

以下是使用Kryo的代碼示例，我們隻要設定序列化類，再注冊要序列化的自定義類型即可（比如算子函數中使用到的外部變量類型、作為RDD泛型類型的自定義類型等）：

// 建立SparkConf對象。
val conf = new SparkConf().setMaster(...).setAppName(...)
// 設定序列化器為KryoSerializer。
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
// 注冊要序列化的自定義類型。
conf.registerKryoClasses(Array(classOf[Studnet], classOf[School]))
           

告訴大家一個開發小技巧：

開發Spark代碼的時候，什麼都不用想，先把序列化器注冊上：

conf.set(“spark.serializer”, “org.apache.spark.serializer.KryoSerializer”)

1.9 優化資料結構

Java中，有三種類型比較耗費記憶體：

• 對象，每個Java對象都有對象頭、引用等額外的資訊，是以比較占用記憶體空間。
• 字元串，每個字元串内部都有一個字元數組以及長度等額外資訊。
• 集合類型，比如HashMap、LinkedList等，因為集合類型内部通常會使用一些内部類來封裝集合元素，比如Map.Entry。

是以Spark官方建議，在Spark編碼實作中，特别是對于算子函數中的代碼，盡量不要使用上述三種資料結構，盡量使用字元串替代對象，使用原始類型（比如Int、Long）替代字元串，使用數組替代集合類型，這樣盡可能地減少記憶體占用，進而降低GC頻率，提升性能。

但是在筆者的編碼實踐中發現，要做到該原則其實并不容易。因為我們同時要考慮到代碼的可維護性，如果一個代碼中，完全沒有任何對象抽象，全部是字元串拼接的方式，那麼對于後續的代碼維護和修改，無疑是一場巨大的災難。同理，如果所有操作都基于數組實作，而不使用HashMap、LinkedList等集合類型，那麼對于我們的編碼難度以及代碼可維護性，也是一個極大的挑戰。是以筆者建議，在可能以及合适的情況下，使用占用記憶體較少的資料結構，但是前提是要保證代碼的可維護性。

總的來說：

（1） 能用json字元串的不要用對象表示，因為對象頭額外占16個位元組

（2） 能不用字元串就不用用字元串，因為字元串額外占40個位元組，比如，能用1 就不要用”1”

（3） 盡量用屬組代替集合類型

（4） 當然不要為了性能好而性能好，我們還是要兼顧代碼的可讀性和開發效率。

1.10 盡可能資料本地化

程序本地化級别：

1.PROCESS_LOCAL：程序本地化，

代碼和資料在同一個程序中，也就是在同一個executor中；計算資料的task由executor執行，資料在executor的BlockManager中；性能最好.

2.NODE_LOCAL：

節點本地化代碼和資料在同一個節點中；比如說，資料作為一個HDFS block塊，就在節點上，而task在節點上某個executor中運作；或者是，資料和task在一個節點上的不同executor中；資料需要在程序間進行傳輸

3.NO_PREF

對于task來說，資料從哪裡擷取都一樣，沒有好壞之分

4.RACK_LOCAL：機架本地化

資料和task在一個機架的兩個節點上；資料需要通過網絡在節點之間進行傳輸

5.ANY

資料和task可能在叢集中的任何地方，而且不在一個機架中，性能最差

我們送出任務後有Spark任務的監控界面，大家一定要利用好這個界面，Spark的界面是做得很好的。比如我們看到這個task的資料本地性是NODE_LOCAL說明是極好的，但是如果有你的task任務的資料本地性較差，可以嘗試如下調優：

如何調優:

spark.locality.wait 預設值是3s 這個代表的意思是，task任務配置設定的時候，先是按照_PROCESS_LOCAL的這種方式去配置設定task的，但是如果PROCESS_LOCAL這個不滿足，那麼預設就等3秒，看能不能按照這級别去配置設定，但是如果等了3秒也實作不了。那麼就按NODE_LOCAL 這個級别去配置設定，以此類推，每次都是等三秒。但是我們知道，如果想代碼運作速度快，那麼就盡可能的讓task配置設定在PROCESS_LOCAL 和 NODE_LOCAL 級别，是以調優的時候，就讓task 在這兩種級别的時候多等一會兒，這樣盡可能的把任務配置設定到這兩個級别。是以預設3秒就有點少了。

spark.locality.wait.process 30s

spark.locality.wait.node 30s

在這兩個級别的時候多等一會兒

2. 基于Spark記憶體模型調優（50分鐘）

概述

我們使用spark-submit送出一個Spark作業之後，這個作業就會啟動一個對應的Driver程序。根據你使用的部署模式（deploy-mode）不同，Driver程序可能在本地啟動，也可能在叢集中某個工作節點上啟動。Driver程序本身會根據我們設定的參數，占有一定數量的記憶體和CPU core。而Driver程序要做的第一件事情，就是向叢集管理器（可以是Spark Standalone叢集，也可以是其他的資源管理叢集，比如我們公司使用的是YARN作為資源管理叢集）申請運作Spark作業需要使用的資源，這裡的資源指的就是Executor程序。YARN叢集管理器會根據我們為Spark作業設定的資源參數，在各個工作節點上，啟動一定數量的Executor程序，每個Executor程序都占有一定數量的記憶體和CPU core。

在申請到了作業執行所需的資源之後，Driver程序就會開始排程和執行我們編寫的作業代碼了。Driver程序會将我們編寫的Spark作業代碼分拆為多個stage，每個stage執行一部分代碼片段，并為每個stage建立一批task，然後将這些task配置設定到各個Executor程序中執行。task是最小的計算單元，負責執行一模一樣的計算邏輯（也就是我們自己編寫的某個代碼片段），隻是每個task處理的資料不同而已。一個stage的所有task都執行完畢之後，會在各個節點本地的磁盤檔案中寫入計算中間結果，然後Driver就會排程運作下一個stage。下一個stage的task的輸入資料就是上一個stage輸出的中間結果。如此循環往複，直到将我們自己編寫的代碼邏輯全部執行完，并且計算完所有的資料，得到我們想要的結果為止。

Spark是根據shuffle類算子來進行stage的劃分。如果我們的代碼中執行了某個shuffle類算子（比如reduceByKey、join等），那麼就會在該算子處，劃分出一個stage界限來。可以大緻了解為，shuffle算子執行之前的代碼會被劃分為一個stage，shuffle算子執行以及之後的代碼會被劃分為下一個stage。是以一個stage剛開始執行的時候，它的每個task可能都會從上一個stage的task所在的節點，去通過網絡傳輸拉取需要自己處理的所有key，然後對拉取到的所有相同的key使用我們自己編寫的算子函數執行聚合操作（比如reduceByKey()算子接收的函數）。這個過程就是shuffle。

當我們在代碼中執行了cache/persist等持久化操作時，根據我們選擇的持久化級别的不同，每個task計算出來的資料也會儲存到Executor程序的記憶體或者所在節點的磁盤檔案中。

是以Executor的記憶體主要分為三塊：第一塊是讓task執行我們自己編寫的代碼時使用；第二塊是讓task通過shuffle過程拉取了上一個stage的task的輸出後，進行聚合等操作時使用（execution記憶體）；第三塊是讓RDD持久化時使用（storage記憶體）。

task的執行速度是跟每個Executor程序的CPU core數量有直接關系的。一個CPU core同一時間隻能執行一個線程。而每個Executor程序上配置設定到的多個task，都是以每個task一條線程的方式，多線程并發運作的。如果CPU core數量比較充足，而且配置設定到的task數量比較合理，那麼通常來說，可以比較快速和高效地執行完這些task線程。

Spark的運作流程

Spark的Stage的劃分流程

為了能把這個資源調優和jvm調優講清楚，我們先講Spark的記憶體模型。

首先我們得明白一個事，我們平時送出的任務設定的—executor-memory 這是值到底是怎麼回事？請看下圖。

在2016年spark1.6版本以前 spark的executor使用的靜态記憶體模型，但是在spark1.6開始，多增加了一個統一記憶體模型。通過spark.memory.useLegacyMode 這個參數去配置。預設這個值是false，帶表用的是新的動态記憶體模型，如果想用以前的靜态記憶體模型，那麼就要把這個值改為true。

2.1 靜态記憶體模型：

實際上就是把我們的一個executor分成了三部分，一部分是Storage記憶體區域，一部分是execution區域，還有一部分是其他區域。如果使用的靜态記憶體模型，那麼用這幾個參數去控制：

spark.storage.memoryFraction：預設0.6

spark.shuffle.memoryFraction：預設0.2

是以第三部分就是0.2

如果我們cache資料量比較大，或者是我們的廣播變量比較大，那我們就把spark.storage.memoryFraction這個值調大一點。但是如果我們代碼裡面沒有廣播變量，也沒有cache，shuffle又比較多，那我們要把spark.shuffle.memoryFraction 這值調大。

靜态記憶體模型的缺點：

我們配置好了Storage記憶體區域和execution區域後，我們的一個任務假設execution記憶體不夠用了，但是它的Storage記憶體區域是空閑的，兩個之間不能互相借用，不夠靈活，是以才出來我們新的統一記憶體模型。

2.2 統一記憶體模型：

動态記憶體模型先是預留了300m記憶體，防止記憶體溢出。動态記憶體模型把整體記憶體分成了兩部分，由這個參數表示spark.memory.fraction 這個指的預設值是0.6 代表另外的一部分是0.4然後spark.memory.fraction 這部分又劃分成為兩個小部分。這兩小部分共占整體記憶體的0.6 .這兩部分其實就是：Storage記憶體和execution記憶體。由spark.memory.storageFraction 這個參數去調配，因為兩個共占0.6。如果spark.memory.storageFraction這個值配的是0.5那說明這0.6裡面 storage占了0.5，也就是execution占了0.1 。

統一記憶體模型有什麼特點呢？

Storage記憶體和execution記憶體 可以互相借用。不用像靜态記憶體模型那樣死闆，但是是有規則的：

比如場景一：Execution使用的時候發現記憶體不夠了，然後就會把storage的記憶體裡的資料驅逐到磁盤上。

比如場景二：一開始execution的記憶體使用得不多，但是storage使用的記憶體多，是以storage就借用了execution的記憶體，但是後來execution也要需要記憶體了，這個時候就會把storage的記憶體裡的資料寫到磁盤上，騰出記憶體空間。

為什麼受傷的都是storage呢？

是因為execution裡面的資料是馬上就要用的，而storage裡的資料不一定馬上就要用。

2.3 資源調優

了解完了Spark作業運作的基本原理之後，對資源相關的參數就容易了解了。所謂的Spark資源參數調優，其實主要就是對Spark運作過程中各個使用資源的地方，通過調節各種參數，來優化資源使用的效率，進而提升Spark作業的執行性能。以下參數就是Spark中主要的資源參數，每個參數都對應着作業運作原理中的某個部分，我們同時也給出了一個調優的參考值。

num-executors

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

executor-memory

• 參數說明：該參數用于設定每個Executor程序的記憶體。Executor記憶體的大小，很多時候直接決定了Spark作業的性能，而且跟常見的JVM OOM異常，也有直接的關聯。
• 參數調優建議：每個Executor程序的記憶體設定4G8G較為合适。但是這隻是一個參考值，具體的設定還是得根據不同部門的資源隊列來定。可以看看自己團隊的資源隊列的最大記憶體限制是多少，num-executors乘以executor-memory，是不能超過隊列的最大記憶體量的。此外，如果你是跟團隊裡其他人共享這個資源隊列，那麼申請的記憶體量最好不要超過資源隊列最大總記憶體的1/31/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（Spark1.6之前的參數）

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

spark.shuffle.memoryFraction（Spark1.6之前的參數）

• 參數說明：該參數用于設定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情況），同時參考本篇文章中給出的原理以及調優建議，合理地設定上述參數。

以下是一份spark-submit指令的示例，大家可以參考一下，并根據自己的實際情況進行調節：

./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 \
           

2.4 個人經驗

java.lang.OutOfMemoryError
ExecutorLostFailure
Executor exit code 為143
executor lost
hearbeat time out
shuffle file lost

如果遇到以上問題，很有可能就是記憶體除了問題，可以先嘗試增加記憶體。如果還是解決不了，那麼請聽下一次資料傾斜調優的課。