因為Spark是記憶體當中的計算框架，叢集中的任何資源都會讓它處於瓶頸，CPU、記憶體、網路頻寬。通常，記憶體足夠的情況之下，網路頻寬是瓶頸，這時我們就需要進行一些調優，比如用一種序列化的方式來儲存RDD來減少記憶體使用，這邊文章就講兩種方式，資料序列化和記憶體調優，接下來我們會分幾個主題來談論這個調優問題。

1、資料序列化

（1） Spark預設是使用Java的ObjectOutputStream框架，它支援所有的繼承於java.io.Serializable序列化,如果想要進行調優的話，可以通過繼承java.io.Externalizable。這種格式比較大，而且速度慢。

（2）Spark還支援這種方式Kryo serialization，它的速度快，而且壓縮比高於Java的序列化，但是它不支援所有的Serializable格式，並且需要在程式裡面註冊。它需要在例項化

import com.esotericsoftware.kryo.Kryo
import org.apache.spark.serializer.KryoRegistrator

class MyRegistrator extends KryoRegistrator {
  override def registerClasses(kryo: Kryo) {
    kryo.register(classOf[MyClass1])
    kryo.register(classOf[MyClass2])
  }
}

// Make sure to set these properties *before* creating a SparkContext!
System.setProperty("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
System.setProperty("spark.kryo.registrator", "mypackage.MyRegistrator")
val sc = new SparkContext(...)
 

如果物件很大，需要設定這個引數spark.kryoserializer.buffer.mb，預設是2。

想了解更多關於這個格式的，可以檢視這個網址https://github.com/EsotericSoftware/kryo

2、記憶體調化

這裡面需要考慮3點，物件使用的記憶體、訪問這些物件的開銷、垃圾回收器的管理開銷。

通常，物件訪問的速度都很快，但是需要2-5x的空間來儲存，因為下面的原因：

1）每一個獨立的Java物件，都有一個16位元組的“object header”和關於這個物件的資訊，比如指標。

2）Java String型別有40位元組的“object header”，然後因為Unicode，每個字元要儲存2個位元組，這樣10個字元要消耗掉大概60個位元組。

3）普通的容器類，比如HashMap和LinkedList，它們採用的是鏈式的資料結構，它需要封裝每個實體，不僅需要頭資訊，還要有個指標指向下一個實體。

4）原始容器型別通常儲存它們為裝箱型別，比如java.lang.Integer。

下面我們就來討論如何確定這些物件的記憶體開銷並且如何進行調優，比如改變資料結構或者序列化儲存資料。下面我們講談論如何調優Spark的Cache大小以及Java的垃圾回收器。

（1）確定記憶體使用情況

首先我們要確定記憶體使用情況，確定資料集的記憶體使用情況，最好的方法就是建立一個RDD，然後快取它，然後檢視日誌，日誌會記錄出來它的每個分片使用的大小，然後我們可以找個這些分片的大小計算出總大小，如下:

INFO BlockManagerMasterActor: Added rdd_0_1 in memory on mbk.local:50311 (size: 717.5 KB, free: 332.3 MB)

（2）資料結構調優

  1) 優先使用陣列和原生型別來替代容器類，或者使用fastutil找個包提供的容器型別，fastutil的官方連結是http://fastutil.di.unimi.it/。

  2）避免大量的小物件的巢狀結構。

  3）使用數字的ID來表示，而不是使用字串的ID。

  4）如果記憶體小於32GB，設定JVM引數 -XX:+UseCompressedOops為4個位元組而不是8個位元組；在Java7或者之後的，嘗試使用-XX:+UseCompressedStrings儲存ASCII字串8個位元一個字元。這些引數可以新增到spark-env.sh，根據我的觀察，應該是設定到SPARK_JAVA_OPTS這個引數上。

（3）序列化RDD儲存

　強烈建議使用Kryo進行序列化，這也是降低記憶體使用最簡單的方式。

（4）垃圾回收器調優

　當我們只使用一次RDD的時候，不會存在這方面的問題。當java需要清除舊的物件給新的物件騰出空間的時候，它需要遍歷所有物件，然後找出那些沒有使用的。這裡最中要的一點是記住，垃圾回收器的代價是和它裡面的物件的數量相關的。檢視GC是不是一個問題，第一件事就是使用序列化的快取方式。

  GC還可以出現的問題就是執行任務所需要的記憶體大小，下面我們講討論如何控制分配給RDD快取的空間大小來減輕這個問題。

  1）確定GC的影響

  新增這些引數到-verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStamps到SPARK_JAVA_OPTS這個引數，讓它出書GC的資訊，然後執行任務。

  2）快取大小調優

　影響GC的一個重要配置引數是分配給快取RDD的記憶體大小，Spark預設是使用 66%的可配置記憶體大小（通過spark.executor.memory or SPARK_MEM來配置）來儲存RDD，也即是說，只有33%是給任務執行過程當中執行過程當中建立的物件的。

　當你的程式慢下來，你發現GC很頻繁或記憶體不夠等現象，降低它的值會起到一些效果，我們可以通過這個引數System.setProperty("spark.storage.memoryFraction", "0.5")來達到這個效果。

　3）高階記憶體調優

　　java的堆記憶體是分為兩個區間，Young和Old，Young是用來儲存短生命週期的物件，Old是用來儲存長生命週期的物件。Young又可以進一步細分為 [Eden, Survivor1, Survivor2]。 一個簡單的垃圾過程可以描述為：當Eden滿的時候，一個簡單的GC會執行在Eden和依賴它的物件，Survivor1被複制到Survivor2。Survivor區域進行了交換。如果一個物件足夠老或者Survivor2滿了，它就會被移到Old區。當Old區也滿的時候，一個完整的GC就會觸發。

Spark裡面的GC調優目標是確保RDD儲存在Old區間，並且Young區有足夠的空間去儲存那些短生命週期的物件。這樣可以減少完全的GC去回收那些任務執行中的臨時物件。 下面的的這些步驟可能是有用的：

　　1）檢查GC的統計資訊，檢視在任務執行完成之前是不是執行過多次的GC，這意味著記憶體不足以執行任務。

      2）當Old區快滿的時候，我們可以通過調整這個引數spark.storage.memoryFraction來減少快取使用的記憶體量，少快取一點物件比拖慢作業執行更好一些。

      3）當發生了很多次小的GC，而不是重要的GC時候，我們可以考慮多分配點記憶體給Eden，假設一個任務需要使用E大小的記憶體，我們可以分配給Eden的記憶體大小為： -Xmn=4/3*E，這個大小同樣適用於survivor區間。

　　4）當從HDFS上讀取資料的時候，任務的所需內容可以估計為block的大小，一個反壓縮的快是2-3倍的大小，我們考慮用3-4個任務來執行，這樣我們可以考慮設定Eden的大小為4*3*64MB。

3、其它的考慮

（1）並行的水平

  建議是1個CPU核心2-3個任務，可以通過程式的函式的時候傳入numPartitions引數，或者通過系統變數spark.default.parallelism來改變。

（2）Reduce任務的記憶體使用情況

  有時候出現OutOfMemoryError並不是因為RDD太大記憶體裝不下，而是因為執行Reduce任務執行的groupByKey的結果太大。Spark的shuffle操作（sortByKey, groupByKey, reduceByKey, join, etc）它會為每一個任務建立一個hash表來執行grouping操作，簡答的處理方式就是增加並行水平，這樣每個任務的輸入集變小。Spark能夠支援每個任務200ms的速度，因為它在所有任務共享了JVMs，減小了釋出任務的開銷，所有可以安全的增加並行水平超過核心數。

 （3）使用broadcast儲存大的變數

　使用Spark裡面的broadcast的變數來儲存大的變數可以大大減少每個序列化任務的大小和叢集釋出任務的開銷。任務大物件的任務都可以考慮使用broadcast變數，Spark在master上會列印每個序列化任務的大小，當大小超過20KB的時候，可以考慮調優。

4、總結

　　這裡簡短的指出了我們調優的時候需要注意的一些重要的點，通常我們把序列化方式調整為Kryo並且快取方式改為序列化儲存方式就可以解決大部分的問題了。