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的錯誤
  • 在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優化，資料序列化等。

5.MaxCompute Spark 相關文章

MaxCompute Spark 使用和常見問題 Spark On MaxCompute如何通路Phonix資料 如何使用MaxCompute Spark讀寫阿裡雲Hbase Spark在MaxCompute的運作方式 Maxcompute Spark作業管控利器—Cupid Console MaxCompute Spark與Spark SQL對比分析及使用注意事項

更多關于大資料計算、雲數倉技術交流，歡迎掃碼加入 “MaxCompute開發者社群” 釘釘群