Spark性能調優-RDD算子調優篇

RDD算子調優

1. RDD複用

在對RDD進行算子時，要避免相同的算子和計算邏輯之下對RDD進行重複的計算，如下圖所示：

對上圖中的RDD計算架構進行修改，得到如下圖所示的優化結果：

2. 盡早filter

擷取到初始RDD後，應該考慮盡早地過濾掉不需要的資料，進而減少對記憶體的占用，進而提升Spark作業的運作效率。

本文首發于公衆号：五分鐘學大資料，歡迎圍觀

1. 讀取大量小檔案-用wholeTextFiles

當我們将一個文本檔案讀取為 RDD 時，輸入的每一行都會成為RDD的一個元素。

也可以将多個完整的文本檔案一次性讀取為一個pairRDD，其中鍵是檔案名，值是檔案内容。

val input:RDD[String] = sc.textFile("dir/*.log") 
           

如果傳遞目錄，則将目錄下的所有檔案讀取作為RDD。檔案路徑支援通配符。

但是這樣對于大量的小檔案讀取效率并不高，應該使用 wholeTextFiles

傳回值為RDD[(String, String)]，其中Key是檔案的名稱，Value是檔案的内容。

def wholeTextFiles(path: String, minPartitions: Int = defaultMinPartitions): RDD[(String, String)])
           

wholeTextFiles讀取小檔案:

val filesRDD: RDD[(String, String)] =
sc.wholeTextFiles("D:\\data\\files", minPartitions = 3)
val linesRDD: RDD[String] = filesRDD.flatMap(_._2.split("\\r\\n"))
val wordsRDD: RDD[String] = linesRDD.flatMap(_.split(" "))
wordsRDD.map((_, 1)).reduceByKey(_ + _).collect().foreach(println)
           

4. mapPartition和foreachPartition

• mapPartitions

map(_…) 表示每一個元素

mapPartitions(_…) 表示每個分區的資料組成的疊代器

普通的map算子對RDD中的每一個元素進行操作，而mapPartitions算子對RDD中每一個分區進行操作。

如果是普通的map算子，假設一個partition有1萬條資料，那麼map算子中的function要執行1萬次，也就是對每個元素進行操作。

如果是mapPartition算子，由于一個task處理一個RDD的partition，那麼一個task隻會執行一次function，function一次接收所有的partition資料，效率比較高。

比如，當要把RDD中的所有資料通過JDBC寫入資料，如果使用map算子，那麼需要對RDD中的每一個元素都建立一個資料庫連接配接，這樣對資源的消耗很大，如果使用mapPartitions算子，那麼針對一個分區的資料，隻需要建立一個資料庫連接配接。

mapPartitions算子也存在一些缺點：對于普通的map操作，一次處理一條資料，如果在處理了2000條資料後記憶體不足，那麼可以将已經處理完的2000條資料從記憶體中垃圾回收掉；但是如果使用mapPartitions算子，但資料量非常大時，function一次處理一個分區的資料，如果一旦記憶體不足，此時無法回收記憶體，就可能會OOM，即記憶體溢出。

是以，mapPartitions算子适用于資料量不是特别大的時候，此時使用mapPartitions算子對性能的提升效果還是不錯的。（當資料量很大的時候，一旦使用mapPartitions算子，就會直接OOM）

在項目中，應該首先估算一下RDD的資料量、每個partition的資料量，以及配置設定給每個Executor的記憶體資源，如果資源允許，可以考慮使用mapPartitions算子代替map。

• foreachPartition

rrd.foreache(_…) 表示每一個元素

rrd.forPartitions(_…) 表示每個分區的資料組成的疊代器

在生産環境中，通常使用foreachPartition算子來完成資料庫的寫入，通過foreachPartition算子的特性，可以優化寫資料庫的性能。

如果使用foreach算子完成資料庫的操作，由于foreach算子是周遊RDD的每條資料，是以，每條資料都會建立一個資料庫連接配接，這是對資源的極大浪費，是以，對于寫資料庫操作，我們應當使用foreachPartition算子。

與mapPartitions算子非常相似，foreachPartition是将RDD的每個分區作為周遊對象，一次處理一個分區的資料，也就是說，如果涉及資料庫的相關操作，一個分區的資料隻需要建立一次資料庫連接配接，如下圖所示：

使用了foreachPartition 算子後，可以獲得以下的性能提升：

1. 對于我們寫的function函數，一次處理一整個分區的資料；
2. 對于一個分區内的資料，建立唯一的資料庫連接配接；
3. 隻需要向資料庫發送一次SQL語句和多組參數；

在生産環境中，全部都會使用foreachPartition算子完成資料庫操作。foreachPartition算子存在一個問題，與mapPartitions算子類似，如果一個分區的資料量特别大，可能會造成OOM，即記憶體溢出。

5. filter+coalesce/repartition(減少分區)

在Spark任務中我們經常會使用filter算子完成RDD中資料的過濾，在任務初始階段，從各個分區中加載到的資料量是相近的，但是一旦進過filter過濾後，每個分區的資料量有可能會存在較大差異，如下圖所示：

根據上圖我們可以發現兩個問題：

1. 每個partition的資料量變小了，如果還按照之前與partition相等的task個數去處理目前資料，有點浪費task的計算資源；
2. 每個partition的資料量不一樣，會導緻後面的每個task處理每個partition資料的時候，每個task要處理的資料量不同，這很有可能導緻資料傾斜問題。

如上圖所示，第二個分區的資料過濾後隻剩100條，而第三個分區的資料過濾後剩下800條，在相同的處理邏輯下，第二個分區對應的task處理的資料量與第三個分區對應的task處理的資料量差距達到了8倍，這也會導緻運作速度可能存在數倍的差距，這也就是資料傾斜問題。

針對上述的兩個問題，我們分别進行分析：

1. 針對第一個問題，既然分區的資料量變小了，我們希望可以對分區資料進行重新配置設定，比如将原來4個分區的資料轉化到2個分區中，這樣隻需要用後面的兩個task進行處理即可，避免了資源的浪費。
2. 針對第二個問題，解決方法和第一個問題的解決方法非常相似，對分區資料重新配置設定，讓每個partition中的資料量差不多，這就避免了資料傾斜問題。

那麼具體應該如何實作上面的解決思路？我們需要coalesce算子。

repartition與coalesce都可以用來進行重分區，其中repartition隻是coalesce接口中shuffle為true的簡易實作，coalesce預設情況下不進行shuffle，但是可以通過參數進行設定。

假設我們希望将原本的分區個數A通過重新分區變為B，那麼有以下幾種情況：

1. A > B（多數分區合并為少數分區）

   • A與B相內插補點不大

     此時使用coalesce即可，無需shuffle過程。

   • A與B相內插補點很大

     此時可以使用coalesce并且不啟用shuffle過程，但是會導緻合并過程性能低下，是以推薦設定coalesce的第二個參數為true，即啟動shuffle過程。

2. A < B（少數分區分解為多數分區）

此時使用repartition即可，如果使用coalesce需要将shuffle設定為true，否則coalesce無效。

我們可以在filter操作之後，使用coalesce算子針對每個partition的資料量各不相同的情況，壓縮partition的數量，而且讓每個partition的資料量盡量均勻緊湊，以便于後面的task進行計算操作，在某種程度上能夠在一定程度上提升性能。

注意：local模式是程序内模拟叢集運作，已經對并行度和分區數量有了一定的内部優化，是以不用去設定并行度和分區數量。

6. 并行度設定

Spark作業中的并行度指各個stage的task的數量。

如果并行度設定不合理而導緻并行度過低，會導緻資源的極大浪費，例如，20個Executor，每個Executor配置設定3個CPU core，而Spark作業有40個task，這樣每個Executor配置設定到的task個數是2個，這就使得每個Executor有一個CPU core空閑，導緻資源的浪費。

理想的并行度設定，應該是讓并行度與資源相比對，簡單來說就是在資源允許的前提下，并行度要設定的盡可能大，達到可以充分利用叢集資源。合理的設定并行度，可以提升整個Spark作業的性能和運作速度。

Spark官方推薦，task數量應該設定為Spark作業總CPU core數量的2~3倍。之是以沒有推薦task數量與CPU core總數相等，是因為task的執行時間不同，有的task執行速度快而有的task執行速度慢，如果task數量與CPU core總數相等，那麼執行快的task執行完成後，會出現CPU core空閑的情況。如果task數量設定為CPU core總數的2~3倍，那麼一個task執行完畢後，CPU core會立刻執行下一個task，降低了資源的浪費，同時提升了Spark作業運作的效率。

Spark作業并行度的設定如下：

val conf = new SparkConf().set("spark.default.parallelism", "500")
           

原則：讓 cpu 的 Core（cpu 核心數） 充分利用起來，

如有100個 Core,那麼并行度可以設定為200~300。

7. repartition/coalesce調節并行度

我們知道 Spark 中有并行度的調節政策，但是，并行度的設定對于Spark SQL是不生效的，使用者設定的并行度隻對于Spark SQL以外的所有Spark的stage生效。

Spark SQL的并行度不允許使用者自己指定，Spark SQL自己會預設根據hive表對應的HDFS檔案的split個數自動設定Spark SQL所在的那個stage的并行度，使用者自己通 spark.default.parallelism 參數指定的并行度，隻會在沒Spark SQL的stage中生效。

由于Spark SQL所在stage的并行度無法手動設定，如果資料量較大，并且此stage中後續的transformation操作有着複雜的業務邏輯，而Spark SQL自動設定的task數量很少，這就意味着每個task要處理為數不少的資料量，然後還要執行非常複雜的處理邏輯，這就可能表現為第一個有Spark SQL的stage速度很慢，而後續的沒有Spark SQL的stage運作速度非常快。

為了解決Spark SQL無法設定并行度和task數量的問題，我們可以使用repartition算子。

repartition 算子使用前後對比圖如下：

Spark SQL這一步的并行度和task數量肯定是沒有辦法去改變了，但是，對于Spark SQL查詢出來的RDD，立即使用repartition算子，去重新進行分區，這樣可以重新分區為多個partition，從repartition之後的RDD操作，由于不再涉及Spark SQL，是以stage的并行度就會等于你手動設定的值，這樣就避免了Spark SQL所在的stage隻能用少量的task去處理大量資料并執行複雜的算法邏輯。使用repartition算子的前後對比如上圖所示。

8. reduceByKey本地預聚合

reduceByKey相較于普通的shuffle操作一個顯著的特點就是會進行map端的本地聚合，map端會先對本地的資料進行combine操作，然後将資料寫入給下個stage的每個task建立的檔案中，也就是在map端，對每一個key對應的value，執行reduceByKey算子函數。

reduceByKey算子的執行過程如下圖所示：

使用reduceByKey對性能的提升如下：

1. 本地聚合後，在map端的資料量變少，減少了磁盤IO，也減少了對磁盤空間的占用；
2. 本地聚合後，下一個stage拉取的資料量變少，減少了網絡傳輸的資料量；
3. 本地聚合後，在reduce端進行資料緩存的記憶體占用減少；
4. 本地聚合後，在reduce端進行聚合的資料量減少。

基于reduceByKey的本地聚合特征，我們應該考慮使用reduceByKey代替其他的shuffle算子，例如groupByKey。

groupByKey與reduceByKey的運作原理如下圖1和圖2所示：

根據上圖可知，groupByKey不會進行map端的聚合，而是将所有map端的資料shuffle到reduce端，然後在reduce端進行資料的聚合操作。由于reduceByKey有map端聚合的特性，使得網絡傳輸的資料量減小，是以效率要明顯高于groupByKey。

9. 使用持久化+checkpoint

Spark持久化在大部分情況下是沒有問題的，但是有時資料可能會丢失，如果資料一旦丢失，就需要對丢失的資料重新進行計算，計算完後再緩存和使用，為了避免資料的丢失，可以選擇對這個RDD進行checkpoint，也就是将資料持久化一份到容錯的檔案系統上（比如HDFS）。

一個RDD緩存并checkpoint後，如果一旦發現緩存丢失，就會優先檢視checkpoint資料存不存在，如果有，就會使用checkpoint資料，而不用重新計算。也即是說，checkpoint可以視為cache的保障機制，如果cache失敗，就使用checkpoint的資料。

使用checkpoint的優點在于提高了Spark作業的可靠性，一旦緩存出現問題，不需要重新計算資料，缺點在于，checkpoint時需要将資料寫入HDFS等檔案系統，對性能的消耗較大。

持久化設定如下：

sc.setCheckpointDir(‘HDFS’)
rdd.cache/persist(memory_and_disk)
rdd.checkpoint
           

10. 使用廣播變量

預設情況下，task中的算子中如果使用了外部的變量，每個task都會擷取一份變量的複本，這就造成了記憶體的極大消耗。一方面，如果後續對RDD進行持久化，可能就無法将RDD資料存入記憶體，隻能寫入磁盤，磁盤IO将會嚴重消耗性能；另一方面，task在建立對象的時候，也許會發現堆記憶體無法存放新建立的對象，這就會導緻頻繁的GC，GC會導緻工作線程停止，進而導緻Spark暫停工作一段時間，嚴重影響Spark性能。

假設目前任務配置了20個Executor，指定500個task，有一個20M的變量被所有task共用，此時會在500個task中産生500個副本，耗費叢集10G的記憶體，如果使用了廣播變量， 那麼每個Executor儲存一個副本，一共消耗400M記憶體，記憶體消耗減少了5倍。

廣播變量在每個Executor儲存一個副本，此Executor的所有task共用此廣播變量，這讓變量産生的副本數量大大減少。

在初始階段，廣播變量隻在Driver中有一份副本。task在運作的時候，想要使用廣播變量中的資料，此時首先會在自己本地的Executor對應的BlockManager中嘗試擷取變量，如果本地沒有，BlockManager就會從Driver或者其他節點的BlockManager上遠端拉取變量的複本，并由本地的BlockManager進行管理；之後此Executor的所有task都會直接從本地的BlockManager中擷取變量。

對于多個Task可能會共用的資料可以廣播到每個Executor上：

val 廣播變量名= sc.broadcast(會被各個Task用到的變量,即需要廣播的變量)

廣播變量名.value//擷取廣播變量
           

11. 使用Kryo序列化

預設情況下，Spark使用Java的序列化機制。Java的序列化機制使用友善，不需要額外的配置，在算子中使用的變量實作Serializable接口即可，但是，Java序列化機制的效率不高，序列化速度慢并且序列化後的資料所占用的空間依然較大。

Spark官方宣稱Kryo序列化機制比Java序列化機制性能提高10倍左右，Spark之是以沒有預設使用Kryo作為序列化類庫，是因為它不支援所有對象的序列化，同時Kryo需要使用者在使用前注冊需要序列化的類型，不夠友善，但從Spark 2.0.0版本開始，簡單類型、簡單類型數組、字元串類型的Shuffling RDDs 已經預設使用Kryo序列化方式了。

Kryo序列化注冊方式的代碼如下：

public class MyKryoRegistrator implements KryoRegistrator{
  @Override
  public void registerClasses(Kryo kryo){
    kryo.register(StartupReportLogs.class);
  }
}
           

配置Kryo序列化方式的代碼如下：

//建立SparkConf對象
val conf = new SparkConf().setMaster(…).setAppName(…)
//使用Kryo序列化庫
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");  
//在Kryo序列化庫中注冊自定義的類集合
conf.set("spark.kryo.registrator", "bigdata.com.MyKryoRegistrator");