Spark 預設採用的是資源預分配的方式。這其實也和按需做資源分配的理念是有衝突的。這篇文章會詳細介紹Spark 動態資源分配原理。

前言

最近在使用Spark Streaming程式時，發現如下幾個問題：

1. 高峰和低峰Spark Streaming每個週期要處理的資料量相差三倍以上,預分配資源會導致低峰的時候資源的大量浪費。
2. Spark Streaming 跑的數量多了後，資源佔用相當可觀。

所以便有了要開發一套針對Spark Streaming 動態資源調整的想法。我在文章最後一個章節給出了一個可能的設計方案。不過要做這件事情，首先我們需要了解現有的Spark 已經實現的 Dynamic Resource Allocation 機制，以及為什麼它無法滿足現有的需求。

入口

在SparkContext 中可以看到這一行：

_executorAllocationManager =
      if (dynamicAllocationEnabled) {
        Some(new ExecutorAllocationManager(this, listenerBus, _conf))
      } else {
        None
      }

通過spark.dynamicAllocation.enabled引數開啟後就會啟動ExecutorAllocationManager。

這裡有我第一個吐槽的點，這麼直接new出來，好歹也做個配置，方便第三方開發個新的元件可以整合進去。但是Spark很多地方都是這麼搞的，完全沒有原來Java社群的風格。

動態調整資源面臨的問題

我們先看看，動態資源調整需要解決哪幾個問題：

1. Cache問題。如果需要移除的Executor含有RDD cache該如何辦？
2. Shuffle問題。 如果需要移除的Executor包含了Shuffle Write先關資料該怎麼辦？
3. 新增和刪除之後都需要告知DAGSchedule進行相關資訊更新。

Cache去掉了重算即可。為了防止資料抖動，預設包含有Cache的Executor是不會被刪除的，因為預設的Idle時間設定的非常大：

private val cachedExecutorIdleTimeoutS = conf.getTimeAsSeconds(  
"spark.dynamicAllocation.cachedExecutorIdleTimeout"
 
, 
s"${Integer.MAX_VALUE}s")

你可以自己設定從而去掉這個限制。

而對於Shuffle,則需要和Yarn整合，需要配置yarn.nodemanager.aux-services。具體配置方式，大家可以Google。這樣Spark Executor就不用儲存Shuffle狀態了。

觸發條件

新增Worker的觸發條件是：

• 有Stage正在執行，並且預估需要的Executors > 現有的

刪除Woker的觸發條件是：

• 一定時間內(預設60s)沒有task執行的Executor

我們看到觸發條件還是比較簡單的。這種簡單就意味著使用者需要根據實際場景，調整各個時間引數，比如到底多久沒有執行task的Executor才需要刪除。

預設檢測時間是100ms:

private val intervalMillis: Long = 100

如何實現Container的新增和釋放

只有ApplicationMaster才能夠向Yarn釋出這些動作。而真正的中控是org.apache.spark.ExecutorAllocationManager,所以他們之間需要建立一個通訊機制。對應的方式是在ApplicationMaster有一個private class AMEndpoint(類，比如刪除釋放容器的動作在裡就有：

  case KillExecutors(executorIds) =>
        logInfo(s"Driver requested to kill executor(s) ${executorIds.mkString(", ")}.")
        Option(allocator) match {
          case Some(a) => executorIds.foreach(a.killExecutor)
          case None => logWarning("Container allocator is not ready to kill executors yet.")
        }
        context.reply(true)

而ExecutorAllocationManager則是引用YarnSchedulerBackend例項，該例項持有ApplicationMaster的 RPC引用

private var amEndpoint: Option[RpcEndpointRef] 

如何獲取排程資訊

要觸發上面描述的操作，就需要任務的排程資訊。這個是通過ExecutorAllocationListener extends SparkListener來完成的。具體是在 ExecutorAllocationMaster的start函式裡，會將該Listener例項新增到SparkContext裡的listenerBus裡，從而實現對DAGSchecude等模組的監聽。機制可以參看這篇文章 Spark ListenerBus 和 MetricsSystem 體系分析。

根據上面的分析，我們至少要知道如下三個資訊：

1. Executor上是否為空,如果為空，就可以標記為Idle.只要超過一定的時間，就可以刪除掉這個Executor.
2. 正在跑的Task有多少
3. 等待排程的Task有多少

這裡是以Stage為區分的。分別以三個變數來表示：

private val stageIdToNumTasks = new mutable.HashMap[Int, Int]
private val stageIdToTaskIndices = new mutable.HashMap[Int, mutable.HashSet[Int]]
private val executorIdToTaskIds = new mutable.HashMap[String, mutable.HashSet[Long]]

名字已經很清楚了。值得說的是stageIdToTaskIndices，其實就是stageId 對應的正在執行的task id 集合。

那麼怎麼計算出等待排程的task數量呢？計算方法如下：

stageIdToNumTasks(stageId) - stageIdToTaskIndices(stageId).size

這些都是動態更新變化的，因為有了監聽器，所以任務那邊有啥變化，這邊都會得到通知。

定時掃描器

有了上面的鋪墊，我們現在進入核心方法：

private def schedule(): Unit = synchronized {
    val now = clock.getTimeMillis

    updateAndSyncNumExecutorsTarget(now)

    removeTimes.retain { case (executorId, expireTime) =>
      val expired = now >= expireTime
      if (expired) {
        initializing = false
        removeExecutor(executorId)
      }
      !expired
    }
  }

該方法會每隔100ms被排程一次。你可以理解為一個監控執行緒。

Executor判定為空閒的機制

只要有一個task結束，就會判定有哪些Executor已經沒有任務了。然後會被加入待移除列表。在放到removeTimes的時候，會把當前時間now + executorIdleTimeoutS * 1000 作為時間戳儲存起來。當排程程序掃描這個到Executor時，會判定時間是不是到了，到了的話就執行實際的remove動作。在這個期間，一旦有task再啟動，並且正好執行在這個Executor上，則又會從removeTimes列表中被移除。 那麼這個Executor就不會被真實的刪除了。

Executor 需要增加的情況

首先，系統會根據下面的公式計算出實際需要的Executors數目：

private def maxNumExecutorsNeeded(): Int = {
    val numRunningOrPendingTasks = listener.totalPendingTasks + listener.totalRunningTasks
    (numRunningOrPendingTasks + tasksPerExecutor - 1) / tasksPerExecutor
  }

接著每個計算週期到了之後，會和當前已經有的Executors數：numExecutorsTarget 進行比較。

1. 如果發現 maxNumExecutorsNeeded < numExecutorsTarget 則會發出取消還有沒有執行的Container申請。並且重置每次申請的容器數為1,也就是numExecutorsToAdd=1

2. 否則如果發現當前時間now >= addTime(addTime 每次會增加一個sustainedSchedulerBacklogTimeoutS ，避免申請容器過於頻繁)，則會進行新容器的申請，如果是第一次，則增加一個(numExecutorsToAdd)，如果是第二次則增加2個以此按倍數類推。直到maxNumExecutorsNeeded <= numExecutorsTarget ,然後就會重置numExecutorsToAdd。

所以我們會發現，我們並不是一次性就申請足夠的資源，而是每隔sustainedSchedulerBacklogTimeoutS次時間，按[1,2,4,8]這種節奏去申請資源的。因為在某個sustainedSchedulerBacklogTimeoutS期間，可能已經有很多工完成了，其實不需要那麼多資源了。而按倍數上升的原因是，防止為了申請到足夠的資源時間花費過長。這是一種權衡。

DRA評價

我們發現，DRA(Dynamic Resource Allocation)涉及到的點還是很多的，雖然邏輯比較簡單，但是和任務排程密切相關，是一個非常動態的過程。這個設計本身也是面向一個通用的排程方式。

我個人建議如果採用了DRA,可以注意如下幾點：

1. 設定一個合理的minExecutors-maxExecutors值
2. 將Executor對應的cpuCore 最好設定為<=3 ，避免Executor數目下降時，等不及新申請到資源，已有的Executor就因為任務過重而導致叢集掛掉。
3. 如果程式中有shuffle,例如(reduce*,groupBy*),建議設定一個合理的並行數，避免殺掉過多的Executors。
4. 對於每個Stage持續時間很短的應用，其實不適合這套機制。這樣會頻繁增加和殺掉Executors，造成系統顛簸。而Yarn對資源的申請處理速度並不快。

Spark Streaming該使用什麼機制動態調整資源

現有的DRA機制其實適合長時的批處理過程中，每個Stage需要的資源量不一樣，並且耗時都比較長。Spark Streaming 可以理解為迴圈的微批處理。而DRA是在每次微批處理起作用，可能還沒等DRA反應過來，這個週期就已經過了。

Spark Streaming需要一個從全域性一天24小時來考慮。每個排程週期的processing time可能更適合作為增減Executors的標準。同時如果發生delay的話，則可以擴大資源申請的速度。並且，因為是週期性的，釋放和新增動作只會發生在一個新的週期的開始，所以他並不會面臨現有 DRA的問題，譬如需要通過額外的方式儲存Shuffle 狀態等。 所以實現起來更加容易。我們可能需要同時監聽StreamingContext的一些資訊。

具體而言：

每個週期檢查上個週期的處理時間 ，設為 preProcessingTime,週期為duration, 一般而言，我們的Spark Streaming程式都會讓preProcessingTime < duration。否則會發生delay。

如果 preProcessingTime > 0.8 * duration,則一次性將資源申請到maxExecutors。

如果preProcessingTime < duration,則應該刪除的Worker為

    removeExecutorNum =  currentExecutors * ((duration -preProcessingTime)/duration - 0.2)

其中0.2 為預留的worker數。如果removeExecutorNum如果<=0 則不進行任何操作。

假設duration =10s, preProcessingTime= 5s, currentExecutors=100，則我們理論上認為 只要保留50%的資源即可。
但是為了防止延時，我們其實額外保留一些20%資源。也就意味著我們刪除30個Executor。 我們並不會一次性將資源都釋放掉。假設我們增加一個新的引數spark.streaming.release.num.duration=5，這個引數意味著我們需要花費5個週期釋放掉這30個Executor的資源。也就是當前這個週期，我們要釋放掉 6個Executor。

接著到下一個週期，重複上面的計算。 直到計算結果 <=0 為止。