本節書摘來自華章社區《深入理解Spark:核心思想與源碼分析》一書中的第3章，第3.10節創建和啟動ExecutorAllocationManager，作者耿嘉安，更多章節內容可以訪問云棲社區“華章社區”公眾號查看

3.10　創建和啟動ExecutorAllocationManager

ExecutorAllocationManager用于對已分配的Executor進行管理，創建和啟動Executor-AllocationManager的代碼如下。 private[spark] val executorAllocationManager: Option[ExecutorAllocationManager] =if (conf.getBoolean("spark.dynamicAllocation.enabled", false)) {Some(new ExecutorAllocationManager(this, listenerBus, conf))} else {None} executorAllocationManager.foreach(_.start())

默認情況下不會創建ExecutorAllocationManager，可以修改屬性spark.dynamicAllocation.enabled為true來創建。ExecutorAllocationManager可以設置動態分配最小Executor數量、動態分配最大Executor數量、每個Executor可以運行的Task數量等配置信息，并對配置信息進行校驗。start方法將ExecutorAllocationListener加入listenerBus中，ExecutorAllocationListener通過監聽listenerBus里的事件，動態添加、刪除Executor。并且通過Thread不斷添加Executor，遍歷Executor，將超時的Executor殺掉并移除。ExecutorAllocationListener的實現與其他SparkListener類似，不再贅述。ExecutorAllocationManager的關鍵代碼見代碼清單3-47。
代碼清單3-47　ExecutorAllocationManager的關鍵代碼

private val intervalMillis: Long = 100 private var clock: Clock = new RealClock private val listener = new ExecutorAllocationListener def start(): Unit = {listenerBus.addListener(listener)startPolling() }private def startPolling(): Unit = {val t = new Thread {override def run(): Unit = {while (true) {try {schedule()} catch {case e: Exception => logError("Exception in dynamic executor allocation thread!", e)}Thread.sleep(intervalMillis)}}}t.setName("spark-dynamic-executor-allocation")t.setDaemon(true)t.start() }

根據3.4.1節的內容，我們知道listenerBus內置了線程listenerThread，此線程不斷從eventQueue中拉出事件對象，調用監聽器的監聽方法。要啟動此線程，需要調用listenerBus的start方法，代碼如下。

listenerBus.start() 《新程序員》：云原生和全面數字化實踐50位技術專家共同創作，文字、視頻、音頻交互閱讀

總結

以上是生活随笔為你收集整理的《深入理解Spark:核心思想与源码分析》——3.10节创建和启动ExecutorAllocationManager...的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。