前言
資源是影響 spark 應用執行效率的一個關鍵因素。Spark 應用中執行任務的組件是 Executor，通過 spark.executor.instances 參數可以設定 Spark 應用的 Executor 數量。在運行過程中，無論 Executor 上是否有任務在執行，它都會被持續占用直到 Spark 應用結束。
在上一篇文章中，我們從動態優化的角度探討了 Spark 3.0 版本中的自適應查詢特性，它主要是在一條 sql 執行過程中不斷優化執行邏輯，選擇更優的執行策略，從而提升性能。本篇我們將從整個 Spark 集群資源的角度探討一個常見問題：資源不足。
在 Spark 集群中，隨著業務的發展，運行的 Spark 應用數量和數據量不斷增加，單純依靠增加資源的優化方式變得越來越不現實。當一個長期運行的 Spark 應用被分配了多個 Executor，但這些 Executor 上卻沒有任務在執行，而此時其他 Spark 應用卻面臨資源緊張，這就導致了資源浪費和調度不合理。
如果每個 Spark 應用的 Executor 數量能夠動態調整就好了。
動態資源分配（Dynamic Resource Allocation）正是為了解決這種情況而設計的。在 Spark 2.4 版本中，kubernetes 上的動態資源分配功能并不完善，但在 Spark 3.0 版本中，Spark on Kubernetes 的功能得到了完善，包括更靈敏的動態分配。我們的 Erda FDP 平臺（Fast Data Platform）從 Spark 2.4 升級到 Spark 3.0，也嘗試了動態資源分配的相關優化。本文將詳細介紹 Spark 3.0 中 Spark on Kubernetes 的動態資源使用情況。
原理
在一個 Spark 應用中，如果某些 Stage 存在數據傾斜，就會導致大量 Executor 處于空閑狀態，造成集群資源的極大浪費。通過動態資源分配策略，空閑的 Executor 如果超過了一定時間，就會被集群回收，并在之后的 Stage 需要時再次請求 Executor。
如下圖所示，在固定 Executor 數量的情況下，Job1 結束和 Job2 開始之間，Executor 處于空閑狀態，造成集群資源的浪費。
開啟動態資源分配后，在 Job1 結束后，Executor1 空閑一段時間后便被回收；在 Job2 需要資源時再申請 Executor2，實現了集群資源的動態管理。
動態分配的原理很容易理解：“按需使用”。當然，一些細節還是需要考慮：
- 何時新增/移除 Executor
- Executor 數量的動態調整范圍
- Executor 的增減頻率
- 在 Spark on Kubernetes 場景下，Executor 的 Pod 銷毀后，它存儲的中間計算數據如何訪問
這些注意點在下面的參數列表中都有相應的說明。
參數一覽
spark.dynamicAllocation.enabled=true #總開關，是否開啟動態資源配置，根據工作負載來衡量是否應該增加或減少executor，默認false spark.dynamicAllocation.shuffleTracking.enabled=true #spark3新增，之前沒有官方支持的on k8s的Dynamic Resouce Allocation。啟用shuffle文件跟蹤，此配置不會回收保存了shuffle數據的executor spark.dynamicAllocation.shuffleTracking.timeout #啟用shuffleTracking時控制保存shuffle數據的executor超時時間，默認使用GC垃圾回收控制釋放。如果有時候GC不及時，配置此參數后，即使executor上存在shuffle數據，也會被回收。暫未配置 spark.dynamicAllocation.minExecutors=1 #動態分配最小executor個數，在啟動時就申請好的，默認0 spark.dynamicAllocation.maxExecutors=10 #動態分配最大executor個數，默認infinity spark.dynamicAllocation.initialExecutors=2 #動態分配初始executor個數默認值=spark.dynamicAllocation.minExecutors spark.dynamicAllocation.executorIdleTimeout=60s #當某個executor空閑超過這個設定值，就會被kill，默認60s spark.dynamicAllocation.cachedExecutorIdleTimeout=240s #當某個緩存數據的executor空閑時間超過這個設定值，就會被kill，默認infinity spark.dynamicAllocation.schedulerBacklogTimeout=3s #任務隊列非空，資源不夠，申請executor的時間間隔，默認1s（第一次申請） spark.dynamicAllocation.sustainedSchedulerBacklogTimeout #同schedulerBacklogTimeout，是申請了新executor之后繼續申請的間隔，默認=schedulerBacklogTimeout（第二次及之后） spark.specution=true #開啟推測執行，對長尾task，會在其他executor上啟動相同task，先運行結束的作為結果
實戰演示
無圖無真相，下面我們將動態資源分配進行簡單演示。
- 配置參數
動態資源分配相關參數配置如下圖所示：
如下圖所示，Spark 應用啟動時的 Executor 個數為 2。因為配置了
spark.dynamicAllocation.initialExecutors=2
運行一段時間后效果如下，executorNum 會遞增，因為空閑的 Executor 被不斷回收，新的 Executor 不斷申請。
- 驗證快慢 SQL 執行
使用 SparkThrfitServer 會遇到的問題是一個數據量很大的 SQL 把所有的資源全占了，導致后面的 SQL 都等待，即使后面的 SQL 只需要幾秒就能完成。我們開啟動態分配策略，再來看 SQL 執行順序。
先提交慢 SQL：
再提交快 SQL：
如下圖所示，開啟動態資源分配后，因為 SparkThrfitServer 可以申請新的 Executor，后面的 SQL 無需等待便可執行。Job7（慢 SQL）還在運行中，后提交的 Job8（快 SQL）已完成。這在一定程度上緩解了資源分配不合理的情況。
- 詳情查看
我們在 SparkWebUI 上可以看到動態分配的整個流程。
登陸 SparkWebUI 頁面，Jobs -> Event Timeline，可以看到 Driver 對整個應用的 Executor 調度。如下圖所示，顯示了每個 Executor 的創建和回收。
同時也能看到此 Executor 的具體創建和回收時間。
在 Executors 標簽頁，我們可以看到所有歷史 Executor 的當前狀態。如下圖所示，之前的 Executor 都已被回收，只有 Executor-31 狀態為 Active。
總結
動態資源分配策略在空閑時釋放 Executor，繁忙時申請 Executor，雖然邏輯比較簡單，但是和任務調度密切相關。它可以防止小數據申請大資源，Executor 空轉的情況。在集群資源緊張，有多個 Spark 應用的場景下，可以開啟動態分配達到資源按需使用的效果。
