MaxCompute Spark 資源使用優化祥解

本文作者：吳數傑阿里雲智能開發工程師

1. 概述

本文主要講解MaxCompute Spark資源調優，目的在於在保證Spark任務正常運行的前提下，指導用戶更好地對Spark作業資源使用進行優化，極大化利用資源，降低成本。

2. Sensor

Sensor提供了一種可視化的方式監控運行中的Spark進程，每個worker（Executor）及master（Driver）都具有各自的狀態監控圖，可以通過Logview中找到入口，如下圖所示：

打開Sensor之後，可以看到下圖提供了Driver/Executor在其生命週期內的CPU和內存的使用情況：

- cpu_plan/mem_plan（藍線）代表了用戶申請的CPU和內存計劃量
- 用戶可以直觀地從cpu_usage圖中看出任務運行中的CPU利用率
- mem_usage代表了任務運行中的內存使用，是mem_rss和page cache兩項之和，詳見下文

Memory Metrics

- mem_rss 代表了進程所佔用了常駐內存，這部分內存也就是Spark任務運行所使用的實際內存，通常需要用戶關注，如果該內存超過用戶申請的內存量，就可能會發生OOM，導致Driver/Executor進程終止。此外，該曲線也可以用於指導用戶進行內存優化，如果實際使用量遠遠小於用戶申請量，則可以減少內存申請，極大化利用資源，降低成本。
- mem_cache（page_cache）用於將磁盤中的數據緩存到內存中，從而減少磁盤I/O操作，通常由系統進行管理，如果物理機內存充足，那麼mem_cache可能會使用很多，用戶可以不必關心該內存的分配和回收。

3. 資源參數調優

（1）Executor Cores

相關參數：spark.executor.cores

- 每個Executor的核數，即每個Executor中的可同時運行的task數目
- Spark任務的最大並行度是num-executors * executor-cores
- Spark任務執行的時候，一個CPU core同一時間最多隻能執行一個Task。如果CPU core數量比較充足，通常來說，可以比較快速和高效地執行完這些Task。同時也要注意，每個Executor的內存是多個Task共享的，如果單個Executor核數太多，內存過少，那麼也很可能發生OOM。

（2）Executor Num

相關參數：spark.executor.instances

- 該參數用於設置Spark作業總共要用多少個Executor進程來執行
- 通常用戶可以根據任務複雜度來決定到底需要申請多少個Executor
- 此外，需要注意，如果出現Executor磁盤空間不足，或者部分Executor OOM的問題，可以通過減少單個Executor的cores數，增加Executor的instances數量來保證任務總體並行度不變，同時降低任務失敗的風險。

（3）Executor Memory

相關參數：spark.executor.memory

- 該參數用於設置每個Executor進程的內存。Executor內存的大小，很多時候直接決定了Spark作業的性能，而且JVM OOM在Executor中更為常見。

相關參數2：spark.executor.memoryOverhead

- 設置申請Executor的堆外內存，主要用於JVM自身，字符串, NIO Buffer等開銷，注意memoryOverhead 這部分內存並不是用來進行計算的，用戶代碼及spark都無法直接操作。
- 如果不設置該值，那麼默認為spark.executor.memory * 0.10，最小為384 MB

Executor 內存不足的表現形式：

- 在Executor的日誌（Logview->某個Worker->StdErr）中出現Cannot allocate memory

- 在任務結束的Logview result的第一行中出現：The job has been killed by "OOM Killer", please check your job's memory usage.
- 在Sensor中發現內存使用率非常高
- 在Executor的日誌中出現java.lang.OutOfMemoryError: Java heap space
- 在Executor的日誌中出現GC overhead limit exceeded
- Spark UI中發現頻繁的GC信息
- 可能出現OOM的間接表現形式：部分Executor出現No route to host: workerd********* / Could not find CoarseGrainedScheduler等錯誤

可能原因及解決方案：

- 限制executor 並行度，將cores 調小：多個同時運行的 Task 會共享一個Executor 的內存，使得單個 Task 可使用的內存減少，調小並行度能緩解內存壓力增加單個Executor內存
- 增加分區數量，減少每個executor負載
- 考慮數據傾斜問題，因為數據傾斜導致某個 task 內存不足，其它 task 內存足夠
- 如果出現了上文所述的Cannot allocate memory或The job has been killed by "OOM Killer", please check your job's memory usage，這種情況通常是由於系統內存不足，可以適當增加一些堆外內存來緩解內存壓力，通常設置spark.executor.memoryOverhead為1g/2g就足夠了

（4）Driver Cores

相關參數spark.driver.cores

- 通常Driver Cores不需要太大，但是如果任務較為複雜（如Stage及Task數量過多）或者Executor數量過多（Driver需要與每個Executor通信並保持心跳），在Sensor中看到Cpu利用率非常高，那麼可能需要適當調大Driver Cores
- 另外要注意，在Yarn-Cluster模式運行Spark任務，不能直接在代碼中設置Driver的資源配置（core/memory），因為在JVM啟動時就需要該參數，因此需要通過--driver-memory命令行選項或在spark-defaults.conf文件/Dataworks配置項中進行設置。

（5）Driver Memory

相關參數1：spark.driver.memory

- 設置申請Driver的堆內內存，與executor類似

相關參數2：spark.driver.maxResultSize

- 代表每個Spark的action（例如collect）的結果總大小的限制，默認為1g。如果總大小超過此限制，作業將被中止，如果該值較高可能會導致Driver發生OOM，因此用戶需要根據作業實際情況設置適當值。

相關參數3：spark.driver.memoryOverhead

- 設置申請Driver的堆外內存，與executor類似

Driver的內存通常不需要太大，如果Driver出現內存不足，通常是由於Driver收集了過多的數據，如果需要使用collect算子將RDD的數據全部拉取到Driver上進行處理，那麼必須確保Driver的內存足夠大。
表現形式：

- Spark應用程序無響應或者直接停止
- 在Driver的日誌（Logview->Master->StdErr）中發現了Driver OutOfMemory的錯誤
- Spark UI中發現頻繁的GC信息
- 在Sensor中發現內存使用率非常高
- 在Driver的日誌中出現Cannot allocate memory

可能原因及解決方案：

- 代碼可能使用了collect操作將過大的數據集收集到Driver節點
- 在代碼創建了過大的數組，或者加載過大的數據集到Driver進程彙總
- SparkContext，DAGScheduler都是運行在Driver端的。對應rdd的Stage切分也是在Driver端運行，如果用戶自己寫的程序有過多的步驟，切分出過多的Stage，這部分信息消耗的是Driver的內存，這個時候就需要調大Driver的內存。有時候如果stage過多，Driver端甚至會有棧溢出

（6）本地磁盤空間

相關參數：spark.hadoop.odps.cupid.disk.driver.device_size：

- 該參數代表為單個Driver或Executor申請的磁盤空間大小，默認值為20g，最大支持100g
- Shuffle數據以及BlockManager溢出的數據均存儲在磁盤上

磁盤空間不足的表現形式：

- 在Executor/Driver的日誌中發現了No space left on device錯誤

解決方案：

- 最簡單的方法是直接增加更多的磁盤空間，調大spark.hadoop.odps.cupid.disk.driver.device_size
- 如果增加到100g之後依然出現該錯誤，可能是由於存在數據傾斜，shuffle或者cache過程中數據集中分佈在某些block，也可能是單個Executor的shuffle數據量確實過大，可以嘗試：

- - 對數據重分區，解決數據傾斜問題
  - 縮小單個Executor的任務併發spark.executor.cores
  - 縮小讀表併發spark.hadoop.odps.input.split.size
  - 增加executor的數量spark.executor.instances

需要注意：

- 同樣由於在JVM啟動前就需要掛載磁盤，因此該參數必須配置在spark-defaults.conf文件或者dataworks的配置項中，不能配置在用戶代碼中
- 此外需要注意該參數的單位為g，不能省略g
- 很多時候由於用戶配置位置有誤或者沒有帶單位g，導致參數實際並沒有生效，任務運行依然失敗

4. 總結

上文主要介紹了MaxCompute Spark在使用過程中可能遇到的資源不足的問題及相應的解決思路，為了能夠最大化利用資源，首先建議按照1: 4的比例來申請單個worker資源，即1 core: 4 gb memory，如果出現OOM，那麼需要查看日誌及Sensor對問題進行初步定位，再進行相應的優化和資源調整。不建議單個Executor Cores 設置過多，通常單個Executor在2-8 core是相對安全的，如果超過8，那麼建議增加instance數量。適當增加堆外內存（為系統預留一些內存資源）也是一個常用的調優方法，通常在實踐中可以解決很多OOM的問題。最後，用戶可以參考官方文檔https://spark.apache.org/docs/2.4.5/tuning.html，包含更多的內存調優技巧，如gc優化，數據序列化等。