摘要：參數說明：當ShuffleManager爲SortShuffl eManager時，如果shuffle read task的數量小於這個閾值（默認是200），則shuffle write過程中不會進行排序操作，而是直接按照未經優化的HashShuffleManager的方式去寫數據，但是最後會將每個task產生的所有臨時磁盤文件都合併成一個文件，並會創建單獨的索引文件。shuffle write操作，我們可以簡單理解爲對pairs RDD中的數據進行分區操作，每個task處理的數據中，相同的key會寫入同一個磁盤文件內。

筆者根據之前的Spark作業開發經驗以及實踐積累，總結出了一套Spark作業的性能優化方案。整套方案主要分爲開發調優、資源調優、數據傾斜調優、shuffle調優幾個部分。開發調優和資源調優是所有Spark作業都需要注意和遵循的一些基本原則，是高性能Spark作業的基礎；數據傾斜調優，主要講解了一套完整的用來解決Spark作業數據傾斜的解決方案；shuffle調優，面向的是對Spark的原理有較深層次掌握和研究的同學，主要講解了如何對Spark作業的shuffle運行過程以及細節進行調優。

Spark開發調優篇

Spark性能優化的第一步，就是要在開發Spark作業的過程中注意和應用一些性能優化的基本原則。開發調優，就是要讓大家瞭解以下一些Spark基本開發原則，包括：RDD lineage設計、算子的合理使用、特殊操作的優化等。在開發過程中，時時刻刻都應該注意以上原則，並將這些原則根據具體的業務以及實際的應用場景，靈活地運用到自己的Spark作業中。

原則一：避免創建重複的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的性能開銷，很明顯是白白浪費掉的。
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(...)

原則二：儘可能複用同一個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><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被多次使用時只被計算一次。

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

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

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

因此對於這種情況，我們的建議是：對多次使用的RDD進行持久化。此時Spark就會根據你的持久化策略，將RDD中的數據保存到內存或者磁盤中。以後每次對這個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，但前提是你的內存必須足夠足夠大，可以綽綽有餘地存放下整個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的級別，必須將所有數據都複製一份副本，併發送到其他節點上，數據複製以及網絡傳輸會導致較大的性能開銷，除非是要求作業的高可用性，否則不建議使用。

原則四：儘量避免使用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作業，可以大大減少性能開銷。

Broadcast與map進行join代碼示例 

// 傳統的join操作會導致shuffle操作。
// 因爲兩個RDD中，相同的key都需要通過網絡拉取到一個節點上，由一個task進行join操作。
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的全量數據。

原則五：使用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數據，都進行了預聚合，然後才傳輸到其他節點上進行全局聚合。

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

除了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來說，性能可能是要高的。

原則七：廣播大變量

有時在開發過程中，會遇到需要在算子函數中使用外部變量的場景（尤其是大變量，比如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...)

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

在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[MyClass1], classOf[MyClass2]))

原則九：優化數據結構

Java中，有三種類型比較耗費內存：

  • 對象，每個Java對象都有對象頭、引用等額外的信息，因此比較佔用內存空間。

  • 字符串，每個字符串內部都有一個字符數組以及長度等額外信息。

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

因此Spark官方建議，在Spark編碼實現中，特別是對於算子函數中的代碼，儘量不要使用上述三種數據結構，儘量使用字符串替代對象，使用原始類型（比如Int、Long）替代字符串，使用數組替代集合類型，這樣儘可能地減少內存佔用，從而降低GC頻率，提升性能。

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

Spark資源調優篇

在開發完Spark作業之後，就該爲作業配置合適的資源了。Spark的資源參數，基本都可以在spark-submit命令中作爲參數設置。很多Spark初學者，通常不知道該設置哪些必要的參數，以及如何設置這些參數，最後就只能胡亂設置，甚至壓根兒不設置。資源參數設置的不合理，可能會導致沒有充分利用集羣資源，作業運行會極其緩慢；或者設置的資源過大，隊列沒有足夠的資源來提供，進而導致各種異常。總之，無論是哪種情況，都會導致Spark作業的運行效率低下，甚至根本無法運行。因此我們必須對Spark作業的資源使用原理有一個清晰的認識，並知道在Spark作業運行過程中，有哪些資源參數是可以設置的，以及如何設置合適的參數值。

一、Spark作業基本運行原理

詳細原理見上圖。我們使用spark-submit提交一個Spark作業之後，這個作業就會啓動一個對應的Driver進程。根據你使用的部署模式（deploy-mode）不同，Driver進程可能在本地啓動，也可能在集羣中某個工作節點上啓動。Driver進程本身會根據我們設置的參數，佔有一定數量的內存和CPU core。而Driver進程要做的第一件事情，就是向集羣管理器申請運行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執行我們自己編寫的代碼時使用，默認是佔Executor總內存的20%；第二塊是讓task通過shuffle過程拉取了上一個stage的task的輸出後，進行聚合等操作時使用，默認也是佔Executor總內存的20%；第三塊是讓RDD持久化時使用，默認佔Executor總內存的60%。

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

以上就是Spark作業的基本運行原理的說明，大家可以結合上圖來理解。理解作業基本原理，是我們進行資源參數調優的基本前提。

二、資源參數調優

瞭解完了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進程的內存設置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情況），同時參考本篇文章中給出的原理以及調優建議，合理地設置上述參數。

三、資源參數參考示例

以下是一份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 \

Spark數據傾斜調優篇

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

一、數據傾斜發生時的現象

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

  • 原本能夠正常執行的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所決定 。

三、如何定位導致數據傾斜的代碼

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

1、某個task執行特別慢的情況

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

如果是用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上，供我們遍歷和打印輸出。

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(_))

通過對單詞計數程序的分析，希望能夠讓大家瞭解最基本的stage劃分的原理，以及stage劃分後shuffle操作是如何在兩個stage的邊界處執行的。然後我們就知道如何快速定位出發生數據傾斜的stage對應代碼的哪一個部分了。比如我們在Spark Web UI或者本地log中發現，stage1的某幾個task執行得特別慢，判定stage1出現了數據傾斜，那麼就可以回到代碼中定位出stage1主要包括了reduceByKey這個shuffle類算子，此時基本就可以確定是由educeByKey算子導致的數據傾斜問題。比如某個單詞出現了100萬次，其他單詞纔出現10次，那麼stage1的某個task就要處理100萬數據，整個stage的速度就會被這個t ask拖慢。

2、某個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的出現次數。

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

五、數據傾斜的解決方案

知道了數據傾斜發生在哪裏之後，通常需要分析一下那個執行了shuffle操作並且導致了數 據傾斜的RDD/Hive表，查看一下其中key的分

解決方案一：使用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中的數據直接通過colle ct算子拉取到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 = rdd1.collect()
// 然後使用Spark的廣播功能，將小RDD的數據轉換成廣播變量，這樣每個Executor就只有一份RDD的數據。
// 可以儘可能節省內存空間，並且減少網絡傳輸性能開銷。
final Broadcast<List<Tuple2<Long, Row>>> rdd1DataBroadcast = sc.broadcast(rdd1Data);

// 對另外一個RDD執行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 {
                // 在算子函數中，通過廣播變量，獲取到本地Executor中的rdd1數據。
                List<Tuple2<Long, Row>> rdd1Data = rdd1DataBroadcast.value();
                // 可以將rdd1的數據轉換爲一個Map，便於後面進行join操作。
                Map<Long, Row> rdd1DataMap = new HashMap<Long, Row>();
                for(Tuple2<Long, Row> 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<String, String>(key, new Tuple2<String, Row>(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操作，選擇一種方案來優化其性能。大家需要對這些方案的思路和原理都透徹理解之後，在實踐中根據各種不同的情況，靈活運用多種方案，來解決自己的數據傾斜問題。

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開始說起。s huffle 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就需要將上一個s tage的計算結果中的所有相同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.consoli dateFiles。該參數默認值爲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機制。

1、普通運行機制

下圖說明了普通的SortShuffleManager的原理。在該模式下，數據會先寫入一個內存數據結構中，此時根據不同的shuffle算子，可能選用不同的數據結構。如果是reduceByKey這種聚合類的shuffle算子，那麼會選用Map數據結構，一邊通過Map進行聚合，一邊寫入內存；如果是join這種普通的shuffle算子，那麼會選用Array數據結構，直接寫入內存。接着，每寫一條數據進入內存數據結構之後，就會判斷一下，是否達到了某個臨界閾值。如果達到臨界閾值的話，那麼就會嘗試將內存數據結構中的數據溢寫到磁盤，然後清空內存數據結構。

在溢寫到磁盤文件之前，會先根據key對內存數據結構中已有的數據進行排序。排序過後，會分批將數據寫入磁盤文件。默認的 batch數量是10000條，也就是說，排序好的數據，會以每批1萬條數據的形式分批寫入磁盤文件。寫入磁盤文件是通過Java的BufferedOutputStream實現的。BufferedOutputStream是Java的緩衝輸出流，首先會將數據緩衝在內存中，當內存緩衝滿溢之後再一次寫入磁盤文件中，這樣可以減少磁盤IO次數，提升性能。

一個task將所有數據寫入內存數據結構的過程中，會發生多次磁盤溢寫操作，也就會產生多個臨時文件。最後會將之前所有的臨時磁盤文件都進行合併，這就是merge過程，此時會將之前所有臨時磁盤文件中的數據讀取出來，然後依次寫入最終的磁盤文件之中。此外，由於一個task就只對應一個磁盤文件，也就意味着該task爲下游stage的task準備的數據都在這一個文件中，因此還會單獨寫一份索引文件，其中標識了下游各個task的數據在文件中的start offset與end offset。

SortShuffleManager由於有一個磁盤文件merge的過程，因此大大減少了文件數量。比如第一個stage有50個task，總共有10個Executor，每個Executor執行5個task，而第二個stage有100個task。由於每個task最終只有一個磁盤文件，因此此時每個Executor上只有5個磁盤文件，所有Executor只有50個磁盤文件。

2、bypass運行機制

下圖說明了bypass SortShuffleManager的原理。bypass運行機制的觸發條件如下：

  • shuffle map task數量小於spark.shuf fle.sort.bypassMergeThreshold參數的值。

  • 不是聚合類的shuffle算子（比如reduceByKey）。

此時task會爲每個下游task都創建一個臨時磁盤文件，並將數據按key進行hash然後根據key的hash值，將key寫入對應的磁盤文件之中。當然，寫入磁盤文件時也是先寫入內存緩衝，緩衝寫滿之後再溢寫到磁盤文件的。最後，同樣會將所有臨時磁盤文件都合併成一個磁盤文件，並創建一個單獨的索引文件。

該過程的磁盤寫機制其實跟未經優化的HashShuffleManager是一模一樣的，因爲都要創建數量驚人的磁盤文件，只是在最後會做一個磁盤文件的合併而已。因此少量的最終磁盤文件，也讓該機制相對未經優化的HashShuffleManager來說，shuffle read的性能會更好。

而該機制與普通SortShuffleManager運行機制的不同在於：第一，磁盤寫機制不同；第二，不會進行排序。也就是說，啓用該機制的最大好處在於，shuffle write過程中，不需要進行數據的排序操作，也就節省掉了這部分的性能開銷。

四、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爲SortShuffl eManager時，如果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%。

寫在最後的話

本文分別講解了開發過程中的優化原則、運行前的資源參數設置調優、運行中的數據傾斜的解決方案、爲了精益求精的shuffle調優。希望大家能夠在閱讀本文之後，記住這些性能調 優的原則以及方案，在Spark作業開發、測試以及運行的過程中多嘗試，只有這樣，我們 才能開發出更優的Spark作業，不斷提升其性能。

往期推薦：

Hive 調優，先掌握這幾種優化模式

視頻 | 58同城HBase平臺及生態建設實踐

相關文章