Spark應用程式開發引數調優深入剖析-Spark商業調優實戰

摘要： 本套系列部落格從真實商業環境抽取案例進行總結和分享，並給出Spark商業應用實戰指導，請持續關注本套部落格。版權宣告：本套Spark商業應用實戰歸作者（秦凱新）所有，禁止轉載，歡迎學習。 Spark商業應用實戰-Spark資料傾斜案例測試及調優準則深入剖析 Spark商...

本套系列部落格從真實商業環境抽取案例進行總結和分享，並給出Spark商業應用實戰指導，請持續關注本套部落格。版權宣告：本套Spark商業應用實戰歸作者（秦凱新）所有，禁止轉載，歡迎學習。

• ofollow,noindex">Spark商業應用實戰-Spark資料傾斜案例測試及調優準則深入剖析
• Spark商業應用實戰-Spark資源排程引數調優深入剖析
• Spark商業應用實戰-Spark應用程式開發引數調優深入剖析
• Spark商業應用實戰-Spark Shuffle 過程引數優化深入剖析

1 Spark內部資源關係

2 Spark執行資源優化配置

./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 \
複製程式碼

3 Spark 運算元調優建議

• 程式開發調優 ：避免建立重複的RDD

  val rdd1 = sc.textFile("hdfs://master01:9000/hello.txt")
  rdd1.map(...)
  val rdd2 = sc.textFile("hdfs://master01:9000/hello.txt")
  rdd2.reduce(...)
  複製程式碼

需要對名為“hello.txt”的HDFS檔案進行一次map操作，再進行一次reduce操作。 也就是說，需要對一份資料執行兩次運算元操作。 錯誤的做法：對於同一份資料執行多次運算元操作時，建立多個RDD。 這裡執行了兩次textFile方法，針對同一個HDFS檔案，建立了兩個RDD出來 ，然後分別對每個RDD都執行了一個運算元操作。 這種情況下，Spark需要從HDFS上兩次載入hello.txt檔案的內容，並建立兩個單獨的RDD； 第二次載入HDFS檔案以及建立RDD的效能開銷，很明顯是白白浪費掉的。

• 程式開發調優 ：儘可能複用同一個RDD

• 錯誤的做法： 有一個<long , String>格式的RDD，即rdd1。 接著由於業務需要，對rdd1執行了一個map操作，建立了一個rdd2， 而rdd2中的資料僅僅是rdd1中的value值而已，也就是說，rdd2是rdd1的子集。

  JavaPairRDD<long , String> rdd1 = ... JavaRDD rdd2 = rdd1.map(...)

分別對rdd1和rdd2執行了不同的運算元操作。

rdd1.reduceByKey(...)
rdd2.map(...)
複製程式碼

• 正確的做法：

rdd2的資料完全就是rdd1的子集而已，卻建立了兩個rdd，並對兩個rdd都執行了一次運算元操作。 此時會因為對rdd1執行map運算元來建立rdd2，而多執行一次運算元操作，進而增加效能開銷。 其實在這種情況下完全可以複用同一個RDD。 我們可以使用rdd1，既做reduceByKey操作，也做map操作。

JavaPairRDD<long , String> rdd1 = ...
rdd1.reduceByKey(...)
rdd1.map(tuple._2...)
複製程式碼

• 程式開發調優 ：對多次使用的RDD進行持久化

  // 正確的做法。
  // cache()方法表示：使用非序列化的方式將RDD中的資料全部嘗試持久化到記憶體中。
  // 此時再對rdd1執行兩次運算元操作時，只有在第一次執行map運算元時，才會將這個rdd1從源頭處計算一次。
  // 第二次執行reduce運算元時，就會直接從記憶體中提取資料進行計算，不會重複計算一個rdd。
  
  val rdd1 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt").cache()
  rdd1.map(...)
  rdd1.reduce(...)
  
  
  正確的做法：
  // 序列化的方式可以減少持久化的資料對記憶體/磁碟的佔用量，進而避免記憶體被持久化資料佔用過多，
  //從而發生頻繁GC。
  
  val rdd1 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt")
  .persist(StorageLevel.MEMORY_AND_DISK_SER)
  rdd1.map(...)
  rdd1.reduce(...)
  複製程式碼

通常不建議使用DISK_ONLY和字尾為_2的級別：因為完全基於磁碟檔案進行資料的讀寫，會導致效能急劇降低，已經網路較大開銷

• 程式開發調優 ：儘量避免使用shuffle類運算元

如果有可能的話，要儘量避免使用shuffle類運算元，最消耗效能的地方就是shuffle過程。

shuffle過程中，各個節點上的相同key都會先寫入本地磁碟檔案中，然後其他節點需要通過網路傳輸拉取各個節點上的磁碟檔案中的相同key。而且相同key都拉取到同一個節點進行聚合操作時，還有可能會因為一個節點上處理的key過多，導致記憶體不夠存放，進而溢寫到磁碟檔案中。因此在shuffle過程中，可能會發生大量的磁碟檔案讀寫的IO操作，以及資料的網路傳輸操作。磁碟IO和網路資料傳輸也是shuffle效能較差的主要原因。

儘可能避免使用reduceByKey、join、distinct、repartition等會進行shuffle的運算元，儘量使用map類的非shuffle運算元。
// 傳統的join操作會導致shuffle操作。
// 因為兩個RDD中，相同的key都需要通過網路拉取到一個節點上，由一個task進行join操作。
val rdd3 = rdd1.join(rdd2)

// Broadcast+map的join操作，不會導致shuffle操作。
// 使用Broadcast將一個數據量較小的RDD作為廣播變數。

// 注意，以上操作，建議僅僅在rdd2的資料量比較少（比如幾百M，或者一兩G）的情況下使用。
// 因為每個Executor的記憶體中，都會駐留一份rdd2的全量資料。
val rdd2Data = rdd2.collect()
val rdd2DataBroadcast = sc.broadcast(rdd2Data)
val rdd3 = rdd1.map(rdd2DataBroadcast...)
複製程式碼

• 程式開發調優 ：使用map-side預聚合的shuffle操作

如果因為業務需要，一定要使用shuffle操作，無法用map類的運算元來替代，那麼儘量使用可以map-side預聚合的運算元 類似於MapReduce中的本地combiner。map-side預聚合之後，每個節點本地就只會有一條相同的key，因為多條相同的key都被聚合起來了。其他節點在拉取所有節點上的相同key時，就會大大減少需要拉取的資料數量，從而也就減少了磁碟IO以及網路傳輸開銷。

建議使用reduceByKey或者aggregateByKey運算元來替代掉groupByKey運算元

• 程式開發調優 ：使用高效能的運算元
• 使用reduceByKey/aggregateByKey替代groupByKey : map-side
• 使用mapPartitions替代普通map : 函式執行頻率
• 使用foreachPartitions替代foreach : 函式執行頻率
• 使用filter之後進行coalesce操作 : filter後對分割槽進行壓縮
• 使用repartitionAndSortWithinPartitions替代repartition與sort類操作

repartitionAndSortWithinPartitions是Spark官網推薦的一個運算元，官方建議，如果需要在repartition重分割槽之後，還要進行排序，建議直接使用repartitionAndSortWithinPartitions運算元

• 程式開發調優 ：廣播大變數

有時在開發過程中，會遇到需要在運算元函式中使用外部變數的場景（尤其是大變數，比如100M以上的大集合），那麼此時就應該使用Spark的廣播（Broadcast）功能來提升效能。 預設情況下，Spark會將該變數複製多個副本，通過網路傳輸到task中，此時每個task都有一個變數副本。如果變數本身比較大的話（比如100M，甚至1G），那麼大量的變數副本在網路中傳輸的效能開銷，以及在各個節點的Executor中佔用過多記憶體導致的頻繁GC，都會極大地影響效能。 廣播後的變數，會保證每個Executor的記憶體中，只駐留一份變數副本，而Executor中的task執行時共享該Executor中的那份變數副本。

• 程式開發調優 ：使用Kryo優化序列化效能
• 1、在運算元函式中使用到外部變數時，該變數會被序列化後進行網路傳輸。
• 2、將自定義的型別作為RDD的泛型型別時（比如JavaRDD，Student是自定義型別），所有自定義型別物件，都會進行序列化。因此這種情況下，也要求自定義的類必須實現Serializable介面。
• 3、使用可序列化的持久化策略時（比如MEMORY_ONLY_SER），Spark會將RDD中的每個partition都序列化成一個大的位元組陣列。

Spark預設使用的是Java的序列化機制，你可以使用Kryo作為序列化類庫，效率要比 Java的序列化機制要高：

// 建立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中，有三種類型比較耗費記憶體：

• 1、物件，每個Java物件都有物件頭、引用等額外的資訊，因此比較佔用記憶體空間。

• 2、字串，每個字串內部都有一個字元陣列以及長度等額外資訊。

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

  Spark官方建議，在Spark編碼實現中，特別是對於運算元函式中的程式碼，儘量不要使用上述三種資料結構，儘量使用字串替代物件，使用原始型別（比如Int、Long）替代字串，使用陣列替代集合型別，這樣儘可能地減少記憶體佔用，從而降低GC頻率，提升效能。