一次JVM GC引發(fā)的Spark調(diào)優(yōu)大全(建議收藏)

一般在我們開發(fā)spark程序的時候，從代碼開發(fā)到上線以及后期的維護(hù)中，在整個過程中都需要涉及到調(diào)優(yōu)的問題，即一開始需要考慮如何把代碼寫的更簡潔高效調(diào)優(yōu)(即代碼優(yōu)化)，待開發(fā)測試完成后，提交任務(wù)時綜合考量該任務(wù)所需的資源(這里涉及到資源調(diào)優(yōu))，上線后是否會出現(xiàn)數(shù)據(jù)傾斜問題（即傾斜調(diào)優(yōu)），以及是否出現(xiàn)頻繁GC問題（這里涉及到GC調(diào)優(yōu)）。

那么本篇通過反推的模式，即通過GC調(diào)優(yōu)進(jìn)行延伸擴(kuò)展，比如出現(xiàn)GC問題是不是可能出現(xiàn)了傾斜？如果沒有出現(xiàn)傾斜，是不是我們給的資源不足？如果資源充足的話，那么是不是我們代碼寫的有問題呢(比如頻繁創(chuàng)建對象等操作)？按照這樣一個思路展開來總結(jié)spark的調(diào)優(yōu)。

JVM的堆、棧、方法區(qū)

如上圖所示，JVM主要由類加載器系統(tǒng)、運(yùn)行時數(shù)據(jù)區(qū)、執(zhí)行引擎和本地接口等組成。

其中運(yùn)行時數(shù)據(jù)區(qū)又由方法區(qū)、堆、Java棧、PC寄存器、本地方法棧組成。

當(dāng)JVM加載一個class文件后，class中的參數(shù)、類型等信息會存儲到方法區(qū)中，程序運(yùn)行時所創(chuàng)建的對象存儲在堆中(堆中不放基本類型和對象引用，只存放對象本身)。當(dāng)每個新線程啟動時，會有自己的程序計數(shù)器(Program Counter Register)和棧，當(dāng)線程調(diào)用方法時，程序計數(shù)器表明下一條執(zhí)行的指令，同時線程棧會存儲線程的方法調(diào)用狀態(tài)(包括局部變量、被調(diào)用的參數(shù)、中間結(jié)果等)。本地方法調(diào)用存儲在獨立的本地方法棧中，或其他獨立的內(nèi)存區(qū)域中。

棧區(qū)由棧楨組成，每個棧楨就是每個調(diào)用的方法的棧，當(dāng)方法調(diào)用結(jié)束后，JVM會彈棧，即拋棄此方法的棧楨。

JVM內(nèi)存劃分

上圖中的劃分是基于JDK7和JDK8，其中有一些變動(主要是永久代的移除)。

JVM內(nèi)存從大體上劃分為三部分：年輕代、老年代、永久代(元空間)

年輕代:所有新生成的對象都會先放到年輕代，年輕代又分為三個區(qū)：Eden區(qū)、兩個Survivor,三者之間的比例為8:1:1。

??Eden區(qū)：大部分對象會在該區(qū)生成，當(dāng)在Eden區(qū)申請空間失敗后，會觸發(fā)Scavenge GC，對Eden區(qū)進(jìn)行GC，清除非存活對象，并把還存活的對象復(fù)制到其中一個Survivor區(qū)中。這里可能會有一個問題，由于默認(rèn)情況下Eden:Survivor1:Survivor2的內(nèi)存占比是8:1:1,如果存活下來的對象是1.5，一個Survivor區(qū)域放不下，那么這個時候就會利用JVM的擔(dān)保機(jī)制，將多余的對象直接放入老年代，會出現(xiàn)老年代囤積一大堆短生命周期的，導(dǎo)致老年代頻繁溢滿，頻繁進(jìn)行Full GC去回收老年代中的對象

??Survivor區(qū):當(dāng)Eden區(qū)滿后，會把還存活的對象復(fù)制到其中一個S區(qū)中，且兩個S區(qū)之間沒有先后順序關(guān)系，同時根據(jù)程序需要Survivor區(qū)是可以配置多個的，這樣可以增加對象在年輕代存在的時間，減少被放到老年代的可能。JVM每次只會使用Eden和其中一塊Survivor區(qū)域來為對象服務(wù)，所以無論什么時候總會有一塊Survivor區(qū)域是空閑的，也就是說年輕代實際可用的內(nèi)存空間為9/10的年輕代空間。

老年代:在年輕代中經(jīng)歷了N次GC之后仍然存活的對象，就會被放到老年代中。該區(qū)域通常存放一些生命周期較長的對象。默認(rèn)情況下，年輕代和老年代的比值為1:2，即老年代占用堆空間大小的2/3，當(dāng)然這個值可以通過-XX:NewRation來調(diào)整

持久代：主要存放靜態(tài)文件、Java類、方法等。在Java 8中該區(qū)域已經(jīng)被移除了，開始使用本地化的內(nèi)存來存放類的元數(shù)據(jù)，也稱之元空間

JVM GC

JVM主要管理兩種類型的內(nèi)存:堆和非堆，簡單來說，堆就是Java代碼可及的內(nèi)存，是留給開發(fā)人員用的，非堆就是JVM留給自己用的。

對于Java的內(nèi)存管理來說其實就是對象的管理，包括對象的分配和釋放。對于GC來說，當(dāng)我們創(chuàng)建對象的時候，GC就開始監(jiān)控這個對象地址、大小以及使用情況，通常GC采用有向圖的方式記錄管理堆中所有對象，通過這種方式來確定哪些對象是可達(dá)的，哪些對象是不可達(dá)的。具體的GC流程如下：

1. 當(dāng)Eden滿了之后，一個小型的GC就會被觸發(fā)(Minor GC)，Eden和Survivor1中幸存仍被使用的對象被復(fù)制到Survivor2。
2. Survivor1和Survivor2區(qū)域進(jìn)行交換，當(dāng)一個對象生存的時間足夠長或者Survivor2滿了之后，就會被轉(zhuǎn)移到Old代
3. 當(dāng)Old空間快滿的時候，這個時候會進(jìn)行Full GC

一般以下幾種情況可能會導(dǎo)致Full GC:

1. 當(dāng)Old空間被寫滿時
2. System.GC()被顯式調(diào)用
3. 上一次GC之后，Heap的各個區(qū)域分配策略動態(tài)變化

以上簡單說明一下jvm相關(guān)知識點，其實spark GC的目的就是要確保老年代只保存長生命周期RDD，同時年輕代的空間又能夠保存短生命周期的對象，這樣就能避免啟動Full GC

Spark對JVM的使用

基于上篇Tungsten on spark 文章的整理，Executor對內(nèi)存的使用主要有以下幾個部分：

1. RDD存儲。當(dāng)對RDD調(diào)用persist或Cache方法時，RDD的partitons會被存儲到內(nèi)存里，那么這塊內(nèi)存也就是Storage內(nèi)存。
2. Shuffle操作。當(dāng)發(fā)生Shuffle時，需要緩沖區(qū)來存儲Shuffle的輸出和聚合的中間結(jié)果,該塊內(nèi)存稱之為Execution內(nèi)存。
3. 用戶代碼。用戶編寫的代碼能夠使用的內(nèi)存空間，也就是其他內(nèi)存(用戶內(nèi)存)

在統(tǒng)一內(nèi)存模式下,整個堆空間分為Spark Memory和User Memory，其中Spark Memory包括Storage Memory和Execution Memory，而且兩者之間可以互相借用空間。

通過spark.memory.fraction參數(shù)來控制Spark Memory在整個堆空間所占的比例

通過spark.memory.storageFraction來設(shè)置Storage Memory占Spark Memory的比例，如果Spark作業(yè)中有較多的RDD持久化操作，該參數(shù)值可以適當(dāng)調(diào)高，保證持久化的數(shù)據(jù)能夠容納在內(nèi)存中，避免內(nèi)存不夠緩存所有的數(shù)據(jù)，只能寫入磁盤中，降低性能。如果Spark作業(yè)中Shuffle類操作比較多，持久化類操作比較少，那么可以適當(dāng)降低該參數(shù)值。

這里給出一個實際的例子來說明一下spark是如何分配內(nèi)存的

/usr/local/spark-current/bin/spark-submit?\
--master?yarn?\
--deploy-mode?client?\
--executor-memory?1G?\
--queue?root.default?\
--class?my.Application?\
--conf?spark.ui.port=4052?\
--conf?spark.port.maxRetries=100?\
--num-executors?2?\
--jars?mongo-spark-connector_2.11-2.3.1.jar?\
App.jar?20201118000000

#?這里配置兩個Executor,每個Executor內(nèi)存給1G

如圖所示，spark申請到了兩個Executor,每個Executor得到的Storage Memory內(nèi)存分別為384.1MB（注意：這里Storage Memory其實就是Storage+Execution的總和內(nèi)存），這里有一個疑惑，我們分配的是每個Executor內(nèi)存為1G,為什么只得到384MB呢？這里給出具體的計算公式：

1. 我們申請為1G內(nèi)存，但是真正拿到內(nèi)存會比這個少，這里涉及到一個Runtime.getRuntime.maxMemory 值的計算（在上篇文章中關(guān)于UnifiedMemoryManager源碼分析中提到過），Runtime.getRuntime.maxMemory對應(yīng)的值才是程序能夠使用的最大內(nèi)存，上面也提到了堆劃分了Eden,Survivor,Tenured區(qū)域，所以該值計算公式為：

   ExecutorMemory = ?Eden + 2 * Survivor + Tenured = 1GB ?= 1073741824 字節(jié)

   systemMemory = Runtime.getRuntime.maxMemory ?= Eden + Survivor + Tenured = 954437176.888888888888889 字節(jié)

   //org.apache.spark.memory.UnifiedMemoryManager（這里討論的還是動態(tài)內(nèi)存模型）
   private?def?getMaxMemory(conf:?SparkConf):?Long?=?{
   ??val?systemMemory?=?conf.getLong("spark.testing.memory",?Runtime.getRuntime.maxMemory)
   ??val?reservedMemory?=?conf.getLong("spark.testing.reservedMemory",
   ????????if?(conf.contains("spark.testing"))?0?else?RESERVED_SYSTEM_MEMORY_BYTES)
   ??val?usableMemory?=?systemMemory?-?reservedMemory
   ??val?memoryFraction?=?conf.getDouble("spark.memory.fraction",?0.6)
   ??
   ??//這里即獲取最大的內(nèi)存值
   ??(usableMemory?*?memoryFraction).toLong
   }
   

2. 基于Spark的動態(tài)內(nèi)存模型設(shè)計，其中有300MB的預(yù)留內(nèi)存，因此剩余可用內(nèi)存為總申請得到的內(nèi)存-預(yù)留內(nèi)存

   reservedMemory = 300MB = 314572800字節(jié)

   usableMemory = systemMemory - reservedMemory = 954437176.888888888888889 - ?314572800 = 639864376.888888888888889字節(jié)

3. Spark Web UI界面上雖然顯示的是Storage Memory，但其實是Execution+Storage內(nèi)存，即該部分占用60%比例

   Storage + Execution = usableMemory * 0.6 = 639864376.888888888888889 * 0.6 = 383918626.133333333333333 字節(jié)

4. 通過第三步驟即可看出實際的內(nèi)存分配情況了，注意：web ui界面得到的結(jié)果計算是除于1000轉(zhuǎn)換得到的值。

GC調(diào)優(yōu)步驟

1. 統(tǒng)計一下GC啟動的頻率和GC使用的總時間，即在spark-submit提交的時候設(shè)置參數(shù)即可如圖所示，這里提高了spark.memory.fraction參數(shù)值，則每個Exectuor實際可用的內(nèi)存也隨之增加了.

   /usr/local/spark-current/bin/spark-submit?\
   --master?yarn?\
   --deploy-mode?client?\
   --executor-memory?1G?\
   --driver-memory?1G?\
   --queue?root.default?\
   --class?my.Application?\
   --conf?spark.ui.port=4052?\
   --conf?spark.port.maxRetries=100?\
   --num-executors?2?\
   --jars?mongo-spark-connector_2.11-2.3.1.jar?\
   --conf?"spark.executor.extraJavaOptions=-XX:+PrintGCDetails?-XX:+PrintGCTimeStamps"?\
   --conf?spark.memory.fraction=0.8?\
   App.jar
   

如圖所示，出現(xiàn)了多次Full GC，首先考慮的是可能配置的Executor內(nèi)存較低，這個時候需要增加Executor Memory來調(diào)節(jié)。

2. 檢查GC日志中是否有過于頻繁的GC。如果一個任務(wù)結(jié)束前，F(xiàn)ull GC執(zhí)行多次，說明老年代空間被占滿了，那么有可能是沒有分配足夠的內(nèi)存。

   1.調(diào)整executor的內(nèi)存，配置參數(shù)executor-memory
   2.調(diào)整老年代所占比例：配置-XX:NewRatio的比例值
   3.降低spark.memory.storageFraction減少用于緩存的空間
   

3. 如果有太多Minor GC，但是Full GC不多，可以給Eden分配更多的內(nèi)存.

   1.比如Eden代的內(nèi)存需求量為E，可以設(shè)置Young代的內(nèi)存為-Xmn=4/3*E,設(shè)置該值也會導(dǎo)致Survivor區(qū)域擴(kuò)張
   2.調(diào)整Eden在年輕代所占的比例，配置-XX:SurvivorRatio的比例值
   

4. 調(diào)整垃圾回收器，通常使用G1GC，即配置-XX:+UseG1GC。當(dāng)Executor的堆空間比較大時，可以提升G1 region size(-XX:G1HeapRegionSize)

   /usr/local/spark-current/bin/spark-submit?\
   --master?yarn?\
   --deploy-mode?client?\
   --executor-memory?1G?\
   --driver-memory?1G?\
   --queue?root.default?\
   --class?my.Application?\
   --conf?spark.ui.port=4052?\
   --conf?spark.port.maxRetries=100?\
   --num-executors?2?\
   --jars?mongo-spark-connector_2.11-2.3.1.jar?\
   --conf?"spark.executor.extraJavaOptions=-XX:+UseG1GC?-XX:G1HeapRegionSize=16M?-XX:+PrintGCDetails?-XX:+PrintGCTimeStamps"?\
   --conf?spark.memory.fraction=0.8?\
   App.jar
   

5. 優(yōu)化代碼，盡量多使用array和string，并使用kyro序列，讓每個Partition都成為字節(jié)數(shù)組

6. 結(jié)合實際的需求，調(diào)整緩存和shuffle計算所占的內(nèi)存比例，即當(dāng)代碼中出現(xiàn)shuffle類操作比較多，而不需要太多緩存的話，則可以適當(dāng)降低Storage Memory所占比例；當(dāng)緩存操作比較多，而Shuffle類操作比較少的話，可以適當(dāng)調(diào)低Execution Memory所占比例。主要是通過spark.storage.storageFraction來控制

7. 開啟堆外內(nèi)存，設(shè)置堆外內(nèi)存大小，這里為了避免OOM

   spark.memory.offHeap.size=4G
   spark.memory.offHeap.enabled=true
   

注意：這里需要說明一下spark.executor.memoryOverhead 和spark.memory.offHeap.size之間的區(qū)別

spark.executor.memoryOverhead是屬于JVM堆外內(nèi)存，用于JVM自身的開銷、內(nèi)部的字符串還有一些本地開銷，spark不會對這塊內(nèi)存進(jìn)行管理。默認(rèn)大小為ExecutorMemory的10%，在spark2.4.5之前，該參數(shù)的值應(yīng)該包含spark.memory.offHeap.size的值。比如spark.memory.offHeap.size配置500M,spark.executor.memoryOverhead默認(rèn)為384M，那么memoryOverhead的值應(yīng)該為884M。

//spark2.4.5之前的
//?Executor?memory?in?MB.
protected?val?executorMemory?=?sparkConf.get(EXECUTOR_MEMORY).toInt

//?Additional?memory?overhead.
protected?val?memoryOverhead:?Int?=?sparkConf.get(EXECUTOR_MEMORY_OVERHEAD).getOrElse(
??math.max((MEMORY_OVERHEAD_FACTOR?*?executorMemory).toInt,?MEMORY_OVERHEAD_MIN)).toInt
protected?val?pysparkWorkerMemory:?Int?=?if?(sparkConf.get(IS_PYTHON_APP))?{
??sparkConf.get(PYSPARK_EXECUTOR_MEMORY).map(_.toInt).getOrElse(0)
}?else?{
??0
}

//?Resource?capability?requested?for?each?executors
private[yarn]?val?resource?=?Resource.newInstance(
??executorMemory?+?memoryOverhead?+?pysparkWorkerMemory,
??executorCores)

//由于memoryOverHead的參數(shù)值理解起來比較困難，而且不易于用戶對每個特定的內(nèi)存區(qū)域進(jìn)行自定義配置，所以在Spark3.0之后進(jìn)行了拆分
//spark3.0之后的資源申請更改為
private[yarn]?val?resource:?Resource?=?{
????val?resource?=?Resource.newInstance(
??????executorMemory?+?executorOffHeapMemory?+?memoryOverhead?+?pysparkWorkerMemory,?executorCores)
????ResourceRequestHelper.setResourceRequests(executorResourceRequests,?resource)
????logDebug(s"Created?resource?capability:?$resource")
????resource
??}

spark.memory.offHeap.size這個參數(shù)指定的內(nèi)存(廣義上是指所有堆外的)，這部分內(nèi)存的申請和釋放是直接進(jìn)行的，不由JVM管理，所以這塊是沒有GC的。

傾斜調(diào)優(yōu)

傾斜部分的調(diào)優(yōu)可以閱讀下面兩篇文章，相對來說已經(jīng)比較全了

Spark數(shù)據(jù)傾斜之騷操作解決方案

數(shù)據(jù)開發(fā)必經(jīng)之路-數(shù)據(jù)傾斜

開發(fā)調(diào)優(yōu)

相信有很多讀者應(yīng)該非常熟悉以下這幾種使用姿勢了，這里就不再重復(fù)詳細(xì)說明了

1. 避免創(chuàng)建重復(fù)的RDD

2. 盡可能復(fù)用同一個RDD

3. 對多次使用的RDD進(jìn)行持久化

4. 盡量避免使用Shuffle算子

5. 使用map-side預(yù)聚合的shuffle操作

6. 使用高性能的算子

   6.1: 使用reduceByKey/aggregateByKey替代groupByKey

   6.2: 使用mapPartitions替代普通map

   6.3: 使用foreachPartitions替代foreach

   6.4: 使用filter之后進(jìn)行coalesce操作

   6.5: 使用repartitionAndSortWithinPartitions替代repartition與sort類操作

7. 廣播大變量

   val?list1?=?...
   val?list1Broadcast?=?sc.broadcast(list1)
   rdd1.map(list1Broadcast...)
   

8. 使用kryo優(yōu)化序列化性能

   //?創(chuàng)建SparkConf對象
   conf.set("spark.serializer",?"org.apache.spark.serializer.KryoSerializer")
   
   //?設(shè)置序列化器為KryoSerializer。
   conf.set("spark.serializer",?"org.apache.spark.serializer.KryoSerializer")
   
   //?注冊要序列化的自定義類型。
   conf.registerKryoClasses(Array(classOf[MyClass1],?classOf[MyClass2]))
   

9. 優(yōu)化數(shù)據(jù)結(jié)構(gòu)，盡量使用字符串代替對象，使用原始類型(如int,Long)代替字符串，使用數(shù)組代替集合類型

資源參數(shù)調(diào)優(yōu)

眾所周知，引起GC主要是內(nèi)存資源問題，一般情況下是不需要對GC進(jìn)行調(diào)優(yōu)的。當(dāng)出現(xiàn)GC問題時，那么就需要思考是哪個環(huán)節(jié)造成內(nèi)存緊張。首先想到的應(yīng)該是配置的內(nèi)存不足，直接加資源，這里整理了一些配置參數(shù)，僅供讀者參考。

--end--

掃描下方二維碼
添加好友，備注【交流】
可私聊交流，也可進(jìn)資源豐富學(xué)習(xí)群