背景

如果在持久化RDD的時候，持久化了大量的資料，那麼Java虛擬機器的垃圾回收就可能成為一個性能瓶頸。因為Java虛擬機器會定期進行垃圾回收，此時就會追蹤所有的java物件，並且在垃圾回收時，找到那些已經不在使用的物件，然後清理舊的物件，來給新的物件騰出記憶體空間。

垃圾回收的效能開銷，是跟記憶體中的物件的數量，成正比的。所以，對於垃圾回收的效能問題，首先要做的就是，使用更高效的資料結構，比如array和string；其次就是在持久化rdd時，使用序列化的持久化級別，而且用Kryo序列化類庫，這樣，每個partition就只是一個物件——一個位元組陣列。

叢集環境

節點
xyz01.aiso.com
硬體環境（VMware 10 虛擬機器）（三臺）
記憶體 6G
CPU 4個1核處理器
硬碟SCSI 500GB
網路 NAT

軟體環境
CentOS release 6.4 (Final) x64
java version “1.8.0_112”
Scala 2.11.8
Hadoop 2.5.0-cdh5.3.6
spark-2.3.0-bin-hadoop2.7
apache-maven-3.3.9

xyz02.aiso.com
硬體環境（VMware 10 虛擬機器）（三臺）
記憶體 4G
CPU 1核處理器
硬碟SCSI 500GB
網路 NAT

軟體環境
CentOS release 6.4 (Final) x64
java version “1.8.0_112”
Scala 2.11.8
Hadoop 2.5.0-cdh5.3.6
spark-2.3.0-bin-hadoop2.7
apache-maven-3.3.9

xyz03.aiso.com
硬體環境（VMware 10 虛擬機器）（三臺）
記憶體 4G
CPU 1核處理器
硬碟SCSI 500GB
網路 NAT

軟體環境
CentOS release 6.4 (Final) x64
java version “1.8.0_112”
Scala 2.11.8
Hadoop 2.5.0-cdh5.3.6
spark-2.3.0-bin-hadoop2.7
apache-maven-3.3.9

叢集規劃
xyz01.aiso.com NameNode DataNode ResourceManager NodeManager QuorumPeerMain SPARK HistoryServer HMaster HRegionServer
xyz02.aiso.com SecondaryNameNode DataNode NodeManager QuorumPeerMain HRegionServer
xyz03.aiso.com DataNode JobHistoryServer NodeManager QuorumPeerMain HRegionServer

yarn資源狀況

日誌格式

123.66.144.115 - - [15/Apr/2018:22:42:25 +0800] “GET /docs/manager-howto.html HTTP/1.1” 200 72733

優化前

spark job程式碼

val logsRDD = initRDD.union(initRDD).union(initRDD)
      .map(line => parseLog(line))
      .filter(x => x != null)
      .cache()

    /**
      * 統計web伺服器所有響應中的最大、最小及平均位元組數
      */
    val contextSize = logsRDD.map { log => log.bytes }
    val maxSize = contextSize.max()
    val minSize = contextSize.min()
    val averageSize = contextSize.reduce(_ + _) / contextSize.count()
    println("響應最大值：" + maxSize + "  最小值：" + minSize + "   平均值：" + averageSize);

    /**
      * 統計各種響應狀態的出現次數
      */
    logsRDD.map(log => (log.status, 1))
      .reduceByKey(_ + _)
      .foreach(result => println(" 響應狀態：" + result._1 + "  計數：" + result._2))

    /**
      * 統計訪問總次數超過1000的前3名的ip
      */
    val result = logsRDD.map(log => (log.ip, 1))
      .reduceByKey(_ + _)
      .filter(result => result._2 > 1000)
      .map(m => (m._2, m._1))
      .top(3)
    for (tuple <- result) {
      println("ip : " + tuple._2 + "  請求次數：" + tuple._1);
    }

    /**
      * 統計請求URI的TopN
      */
    val topN = logsRDD.map { log => (log.request.split(" ")(1), 1) }
      .reduceByKey(_ + _)
      .map(result => (result._2, result._1))
      .sortByKey(false)
      .take(3)
    for (tuple <- topN) {
      println("URI : " + tuple._2 + "  請求頻次：" + tuple._1);
    }

提交指令碼

  /opt/module/apache-2.7-ha/spark-2.3.0-bin-hadoop2.7/bin/spark-submit \
  --master yarn \
  --deploy-mode client \
  --executor-memory 500M \
  --num-executors 2 \
  --executor-cores 2 \
  --driver-memory 1G \
   --conf "spark.executor.extraJavaOptions=-XX:+PrintGCDetails  -verbose:gc -XX:+PrintGCTimeStamps " \
  --queue root.xiaoyuzhou \
   --jars ${SPARK_REPORT_EXPORT_JARS} \
   --class com.spark.learn.job.TomcatLogStat \
  $JOB_HOME/sparkexamples-1.0.jar \
   env=prd  \
   log.level=error \ 
   input.data.path=/data/tomcat_localhost_access_log

執行結果

優化分析

1、使用高效的垃圾回收器

此任務使用 -XX:+UseG1GC前後變化：

GC日誌

使用前

使用後

執行時間的變化
App ID App Name Started Completed Duration Spark User Last Updated Event Log
application_1526440328204_0005 TomcatLogStat 2018-05-16 15:21:54 2018-05-16 15:31:42 9.8 min xiaoyuzhou 2018-05-16 15:31:42 Download
application_1526440328204_0004 TomcatLogStat 2018-05-16 14:41:10 2018-05-16 14:57:42 17 min xiaoyuzhou 2018-05-16 14:57:42

調整JVM記憶體空間大小

降低spark.storage.memoryFraction

Spark中，垃圾回收調優的目標就是，只有真正長時間存活的物件，才能進入老年代，短時間存活的物件，只能呆在年輕代。不能因為某個Survivor區域空間不夠，在Minor GC時，就進入了老年代。從而造成短時間存活的物件，長期呆在老年代中佔據了空間，而且Full GC時要回收大量的短時間存活的物件，導致Full GC速度緩慢。

如果發現，在task執行期間，大量full gc發生了，那麼說明，年輕代的Eden區域，給的空間不夠大。

此時可以執行一些操作來優化垃圾回收行為：

1、包括降低spark.storage.memoryFraction的比例，給年輕代更多的空間，來存放短時間存活的物件；

使用前

使用後

jvm eden區調節

使用前

使用後

1、給Eden區域分配更大的空間，使用-Xmn即可，通常建議給Eden區域，預計大小的4/3；
如果使用的是HDFS檔案，那麼很好估計Eden區域大小，如果每個executor有4個task，然後每個hdfs壓縮塊解壓縮後大小是3倍，此外每個hdfs塊的大小是64M，那麼Eden區域的預計大小就是：4 * 3 * 64MB，然後呢，再通過-Xmn引數，將Eden區域大小設定為4 * 3 * 64 * 4/3。

-XX:SurvivorRatio=4：如果值為4，那麼就是兩個Survivor跟Eden的比例是2:4，也就是說每個Survivor佔據的年輕代的比例是1/6，所以，你其實也可以嘗試調大Survivor區域的大小。
-XX:NewRatio=4：調節新生代和老年代的比例, 表示年輕代與年老代所佔比值為1:4,年輕代佔整個堆疊的1/5
Xms=Xmx 並且設定了Xmn的情況下，該引數不需要進行設定。

調節executor的堆外記憶體

當spark作業中，是不是的報錯，shuffle file cannot find,executro、task lost,out of memory等，可能是堆外記憶體不足，導致executor掛掉，task拉取該executor的資料是無法獲取到，導致以上錯誤，甚至spark作業崩潰。

在spark作業的提交指令碼中，修改spark.yarn.executor.memoryOverhead引數（預設為256M），在指令碼中新增引數

--conf spark.yarn.executor.memoryOverhead=2048 \        調節堆外記憶體
--conf spark.core.connection.ack.wait.timeout=300 \       調節連線時間

調節spark 記憶體管理相關引數

使用前

使用後

瞭解 Spark Shuffle 中的 JVM 記憶體使用空間對一個Spark應用程式的記憶體調優是至關重要的。跟據不同的記憶體控制原理分別對儲存和執行空間進行引數調優：spark.executor.memory, spark.storage.safetyFraction, spark.storage.memoryFraction, spark.storage.unrollFraction, spark.shuffle.memoryFraction, spark.shuffle.safteyFraction。

Spark 1.6 以前的版本是使用固定的記憶體分配策略，把 JVM Heap 中的 90% 分配為安全空間，然後從這90%的安全空間中的 60% 作為儲存空間，例如進行 Persist、Unroll 以及 Broadcast 的資料。然後再把這60％的20％作為支援一些序列化和反序列化的資料工作。其次當程式執行時，JVM Heap 會把其中的 80% 作為執行過程中的安全空間，這80％的其中20％是用來負責 Shuffle 資料傳輸的空間。

Spark 2.0 中推出了聯合記憶體的概念，最主要的改變是儲存和執行的空間可以動態移動。需要注意的是執行比儲存有更大的優先值，當空間不夠時，可以向對方借空間，但前提是對方有足夠的空間或者是 Execution 可以強制把 Storage 一部份空間擠掉。Excution 向 Storage 借空間有兩種方式：第一種方式是 Storage 曾經向 Execution 借了空間，它快取的資料可能是非常的多，當 Execution 需要空間時可以強制拿回來；第二種方式是 Storage Memory 不足 50% 的情況下，Storgae Memory 會很樂意地把剩餘空間借給 Execution。

如果是你的計算比較複雜的情況，使用新型的記憶體管理 (Unified Memory Management) 會取得更好的效率，但是如果說計算的業務邏輯需要更大的快取空間，此時使用老版本的固定記憶體管理 (StaticMemoryManagement) 效果會更好。

總結

jvm的調優是非常複雜和敏感的，儘量調節executor記憶體的比例就可以了。真的到了萬不得已的地方，並且對jvm相關的技術很瞭解，那麼再進行eden區域的調節。總之，JVM調優沒有“銀彈”，結合系統現狀和多嘗試不同的調優策略是找到合適調優方法的唯一途徑。