Spark 性能優化：代碼中經常使用的幾個調整點

前面三篇文章的三種優化方式調整到位以後會讓整個 Spark 做業執行速度有很是明顯的提高。

除此以外咱們還有不少其它性能優化的手段，但在和前面三種方式比較，正常狀況下提高沒有那麼大。

1，使用廣播變量

在 task 執行算子函數運算的時候，若是要用到外部變量，這種時候須要使用廣播變量。由於，若是不使用廣播變量，那麼所使用的外部變量會在每一個 task 裏會得到一份變量的副本，後續傳輸到各個 worker 節點去計算的時候一方面會加大不少的網絡傳輸，另外一方面在 task 計算的時候會佔用太大的內存空間（甚至會用到磁盤空間），讀寫消耗的 IO 也會不少。

使用廣播變量後，初始的時候，只會在 Driver 上有一份副本。當 task 計算的時候，須要使用到廣播變量中的數據時，首先會在本地的 executor 對應的 blockManager 中嘗試獲取，若是這裏沒有，就會從 driver 遠程拉取變量副本（也可能會從距離最近的其它節點上的 blockManager 上拉取數據），並保持在本地的 blockManager 中，此後這個 executor 上的全部 task 會直接使用這份數據。

2，使用 kryo 序列化

Spark 內部默認使用的序列化方式是 Java 的序列化方式。和 kryo 的序列化相比，Java 的序列化方式，速度慢，序列化的數據佔用空間相對仍是很大，因此使用 kryo 序列化在必定程度上提高性能。

使用 kryo 序列化能優化到的幾個地方：算子函數中用到的外部變量，持久化 RDD 時候使用 StorageLevel.MEMORY_ONLY_SER 這個級別時，shuffle 過程。

使用方法：

SparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer").registerKryoClasses(xxxx)
複製代碼

3，使用 fastutil 工具類

fastutil 是擴展了 Java 標準集合框架的類庫。他可以提供更小的內存佔用和存取速度。在 Spark 內使用到集合存儲數據的地方能夠儘可能使用 fastutil 提供的一些集合。

4，調整數據本地化等待時長

Spark 在 Driver 上分配每一個 stage 的 task 去 worker 節點以前，都會計算出每一個 task 要計算的是哪一個 partition 數據。這裏 Spark 會有本身的一套 task 分配算法，優先但願的是 task 恰好分配到其全部處理數據的那個節點，能夠省去數據網絡傳輸的性能消耗。

但，若是那個節點上的計算能力或者其餘資源已經滿了，這時候，Spark 會等待一段時間，若是說在這段時間內這個幾點計算能力或者資源釋放了，可以知足這個 task 的資源需求的話，就會繼續分配到這個節點，若是這段時間內仍是老樣子，那麼久得分配到其它節點上了。

這個等待時間是經過 spark.locality.wait 這個配置參數設置。

5，JVM 調優

a，下降 cache 操做的內存佔比： Spark 集羣的每一個 worker 節點上 executor 都是運行在各自的 JVM 中。這個 JVM 的內存會被劃分紅兩塊，一塊是用來給 RDD 的 cache、persist 的數據作緩存的，另外一塊是存儲各個算子中函數運算中產生的對象的。這個比例默認是 0.6，也就是隻有 40% 的內存空間是給各個算子計算用的，若是計算中產生的數據量過大會頻繁觸發 minor gc，甚至會觸發大量的 full gc，很吃性能。因此在 RDD 緩存使用不多的狀況下能夠調節下這個參數。

調節參數爲：spark.storage.memoryFraction

b，調整 JVM 進程中除了 堆內存之外的內存空間： 有時候，若是你的spark做業處理的數據量特別特別大，幾億數據量。而後spark做業一運行，時不時的報錯，shuffle file cannot find，executor、task lost，out of memory。這時候就要考慮設置這個參數了。

調節參數：--conf spark.yarn.executor.memoryOverhead 這個是配置在 shell 腳本里的。

c，調整鏈接等待時長： 碰到一種狀況，沒有任何規律：某某file。一串file id。uuid（dsfsfd-2342vs—sdf--sdfsd）。not found。file lost。這種狀況頗有可能就是某個 worker 上的 executor 拉取另外節點上的數據因爲長時間沒拉到，超過了超時等待時間，爆出的錯誤。

調節參數：--conf spark.core.connection.ack.wait.timeout 配置在 shell 腳本里的。

關於堆外內存和 JVM 內存分配的東西能夠參考這兩偏文章：

www.ibm.com/developerwo…

blog.csdn.net/baolibin528…

6，shuffle 調優

a，合併 map 端輸出文件： shuffle 階段，前一個 stage 會爲後一個 stage 每一個 task 準備一份 map 端的輸出文件，供後一個 stage 拉取。若是後一個 stage 的 task 數量不少，那麼前一個 stage 產生的 map 輸出文件特別多，這時候須要開啓 map 端文件合併，可以減小不少文件，有助於 shuffle 階段的性能提高。

調節參數：new SparkConf().set("spark.shuffle.consolidateFiles", "true")

b，調整 map 端內存緩衝和 reduce 端的內存佔比： map 端緩衝內存大小默認是 32k，reduce 端用來緩存的內存佔比默認是 0.2。

調節參數爲：spark.shuffle.file.buffer spark.shuffle.memoryFraction

c，HashShuffleManager 和 SortShuffleManager： 參考：my.oschina.net/hblt147/blo…

7，算子調優

MapPartitions提高Map類操做性能、filter事後使用coalesce減小分區數量、foreachPartition優化寫數據庫性能、repartition解決Spark SQL低並行度的性能問、reduceByKey 在 shuffle 操做時會在 map 端進行一次本地 combine，性能比 groupByKey 要好不少，因此能用 reduceByKey 的地方儘可能用 reduceByKey。