Spark性能調優-基礎篇

前言

在大數據計算領域，Spark已經成爲了愈來愈流行、愈來愈受歡迎的計算平臺之一。Spark的功能涵蓋了大數據領域的離線批處理、SQL類處理、流式/實時計算、機器學習、圖計算等各類不一樣類型的計算操做，應用範圍與前景很是普遍。

然而，經過Spark開發出高性能的大數據計算做業，並非那麼簡單的。若是沒有對Spark做業進行合理的調優，Spark做業的執行速度可能會很慢，這樣就徹底體現不出Spark做爲一種快速大數據計算引擎的優點來。所以，想要用好Spark，就必須對其進行合理的性能優化。

Spark的性能調優其實是由不少部分組成的，不是調節幾個參數就能夠立竿見影提高做業性能的。咱們須要根據不一樣的業務場景以及數據狀況，對Spark做業進行綜合性的分析，而後進行多個方面的調節和優化，才能得到最佳性能。

筆者根據以前的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, 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	使用未序列化的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的級別，必須將全部數據都複製一份副本，併發送到其餘節點上，數據複製以及網絡傳輸會致使較大的性能開銷，除非是要求做業的高可用性，不然不建議使用。

原則四：儘可能避免使用shuffle類算子

若是有可能的話，要儘可能避免使用shuffle類算子。由於Spark做業運行過程當中，最消耗性能的地方就是shuffle過程，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相關的算子有優化原則以外，其餘的算子也都有着相應的優化原則。

使用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來講，性能多是要高的。

原則七：廣播大變量

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

注意：這裏的大集合是相對來講的，若是是好比groupby以後的數據通常達到10G或100G上的變量不建議廣播，由於每一個task會有一個變量副本反而會下降spark的性能。

在算子函數中使用到外部變量時，默認狀況下，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-submit命令中做爲參數設置。不少Spark初學者，一般不知道該設置哪些必要的參數，以及如何設置這些參數，最後就只能胡亂設置，甚至壓根兒不設置。資源參數設置的不合理，可能會致使沒有充分利用集羣資源，做業運行會極其緩慢；或者設置的資源過大，隊列沒有足夠的資源來提供，進而致使各類異常。總之，不管是哪一種狀況，都會致使Spark做業的運行效率低下，甚至根本沒法運行。所以咱們必須對Spark做業的資源使用原理有一個清晰的認識，並知道在Spark做業運行過程當中，有哪些資源參數是能夠設置的，以及如何設置合適的參數值。

Spark做業基本運行原理

詳細原理見上圖。咱們使用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執行咱們本身編寫的代碼時使用，默認是佔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 \

轉載自https://tech.meituan.com/spark-tuning-basic.html?utm_source=tool.lu