Spark性能優化的基礎是什么

Spark性能優化的基礎是什么，針對這個問題，這篇文章詳細介紹了相對應的分析和解答，希望可以幫助更多想解決這個問題的小伙伴找到更簡單易行的方法。

前言

在大數據計算領域，Spark已經成為了越來越流行、越來越受歡迎的計算平臺之一。Spark的功能涵蓋了大數據領域的離線批處理、SQL類處理、流式/實時計算、機器學習、圖計算等各種不同類型的計算操作，應用范圍與前景非常廣泛。在美團?大眾點評，已經有很多同學在各種項目中嘗試使用Spark。大多數同學（包括筆者在內），最初開始嘗試使用Spark的原因很簡單，主要就是為了讓大數據計算作業的執行速度更快、性能更高。

然而，通過Spark開發出高性能的大數據計算作業，并不是那么簡單的。如果沒有對Spark作業進行合理的調優，Spark作業的執行速度可能會很慢，這樣就完全體現不出Spark作為一種快速大數據計算引擎的優勢來。因此，想要用好Spark，就必須對其進行合理的性能優化。

Spark的性能調優實際上是由很多部分組成的，不是調節幾個參數就可以立竿見影提升作業性能的。我們需要根據不同的業務場景以及數據情況，對Spark作業進行綜合性的分析，然后進行多個方面的調節和優化，才能獲得最佳性能。

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

作為Spark性能優化的基礎篇，小編主要講解開發調優以及資源調優。

調優概述

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

一個簡單的例子

// 需要對名為“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(...)

一個簡單的例子

// 錯誤的做法。

// 有一個<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進行持久化的代碼示例

// 如果要對一個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()方法而言，我們可以根據不同的業務場景選擇不同的持久化級別。

持久化級別	含義解釋
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計算時還可以使用該數據在其他節點上的副本。如果沒有副本的話，就只能將這些數據從源頭處重新計算一遍了。

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

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

使用reduceByKey/aggregateByKey替代groupByKey

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

使用foreachPartitions替代foreach

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

使用repartitionAndSortWithinPartitions替代repartition與sort類操作

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

廣播大變量的代碼示例

// 以下代碼在算子函數中，使用了外部的變量。
// 此時沒有做任何特殊操作，每個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...)

原則九：優化數據結構

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

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

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

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

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

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

調優概述

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

資源參數調優

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

executor-memory

• 參數說明：該參數用于設置每個Executor進程的內存。Executor內存的大小，很多時候直接決定了Spark作業的性能，而且跟常見的JVM OOM異常，也有直接的關聯。

• 參數調優建議：每個Executor進程的內存設置4G~8G較為合適。但是這只是一個參考值，具體的設置還是得根據不同部門的資源隊列來定。可以看看自己團隊的資源隊列的最大內存限制是多少，num-executors乘以executor-memory，是不能超過隊列的最大內存量的。此外，如果你是跟團隊里其他人共享這個資源隊列，那么申請的內存量最好不要超過資源隊列最大總內存的1/3~1/2，避免你自己的Spark作業占用了隊列所有的資源，導致別的同學的作業無法運行。

driver-memory

• 參數說明：該參數用于設置Driver進程的內存。

• 參數調優建議：Driver的內存通常來說不設置，或者設置1G左右應該就夠了。唯一需要注意的一點是，如果需要使用collect算子將RDD的數據全部拉取到Driver上進行處理，那么必須確保Driver的內存足夠大，否則會出現OOM內存溢出的問題。

spark.storage.memoryFraction

• 參數說明：該參數用于設置RDD持久化數據在Executor內存中能占的比例，默認是0.6。也就是說，默認Executor 60%的內存，可以用來保存持久化的RDD數據。根據你選擇的不同的持久化策略，如果內存不夠時，可能數據就不會持久化，或者數據會寫入磁盤。

• 參數調優建議：如果Spark作業中，有較多的RDD持久化操作，該參數的值可以適當提高一些，保證持久化的數據能夠容納在內存中。避免內存不夠緩存所有的數據，導致數據只能寫入磁盤中，降低了性能。但是如果Spark作業中的shuffle類操作比較多，而持久化操作比較少，那么這個參數的值適當降低一些比較合適。此外，如果發現作業由于頻繁的gc導致運行緩慢（通過spark web ui可以觀察到作業的gc耗時），意味著task執行用戶代碼的內存不夠用，那么同樣建議調低這個參數的值。

資源參數參考示例

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

<h2 id="-" white-space:normal;"> 

關于Spark性能優化的基礎是什么問題的解答就分享到這里了，希望以上內容可以對大家有一定的幫助，如果你還有很多疑惑沒有解開，可以關注億速云行業資訊頻道了解更多相關知識。