# 作業調度 * [概述](#概述) * [跨應用調度](#跨應用調度) * [動態資源分配](#動態資源分配) * [配置和部署](#配置和部署) * [資源分配策略](#資源分配策略) * [請求策略](#請求策略) * [移除策略](#移除策略) * [優雅的關閉Executor(執行器)](#優雅的關閉executor執行器) * [應用內調度](#應用內調度) * [公平調度資源池](#公平調度資源池) * [資源池默認行為](#資源池默認行為) * [配置資源池屬性](#配置資源池屬性) # 概述 Spark 有好幾計算資源調度的方式。首先，回憶一下 [集群模式概述](cluster-overview.html)，每個 Spark 應用(包含一個 SparkContext 實例)中運行了一些其獨占的執行器（executor）進程。集群管理器提供了 Spark 應用之間的資源調度[scheduling across applications](#scheduling-across-applications)。其次，在各個 Spark 應用內部，各個線程可能并發地通過 action 算子提交多個 Spark 作業（job）。如果你的應用服務于網絡請求，那這種情況是很常見的。在 Spark 應用內部（對應同一個 SparkContext）各個作業之間，Spark 默認 FIFO 調度，同時也可以支持公平調度 [fair scheduler](#scheduling-within-an-application)。 # 跨應用調度 如果在集群上運行，每個 Spark 應用都會 SparkContext 獲得一批獨占的執行器 JVM，來運行其任務并存儲數據。如果有多個用戶共享集群，那么會有很多資源分配相關的選項，如何設計還取覺于具體的集群管理器。 對 Spark 所支持的各個集群管理器而言，最簡單的的資源分配，就是靜態劃分。這種方式就意味著，每個 Spark 應用都是設定一個最大可用資源總量，并且該應用在整個生命周期內都會占住這個資源。這種方式在 Spark’s 獨立部署 [standalone](spark-standalone.html) 和 [YARN](running-on-yarn.html)調度，以及 Mesos 粗粒度模式下都可用。[coarse-grained Mesos mode](running-on-mesos.html#mesos-run-modes)。Resource allocation can be configured as follows，based on the cluster type: * **Standalone mode:** 默認情況下，Spark 應用在獨立部署的集群中都會以 FIFO（first-in-first-out）模式順序提交運行，并且每個spark應用都會占用集群中所有可用節點。不過你可以通過設置 spark.cores.max 或者 spark.deploy.defaultCores 來限制單個應用所占用的節點個數。最后，除了可以控制對 CPU 的使用數量之外，還可以通過spark.executor.memory來控制各個應用的內存占用量。 * **Mesos:** 在Mesos中要使用靜態劃分的話，需要將 spark.mesos.coarse 設為true，同樣，你也需要配置 spark.cores.max來控制各個應用的 CPU 總數，以及 spark.executor.memory 來控制各個應用的內存占用。 * **YARN:** 在 YARN 中需要使用 –num-executors 選項來控制 Spark 應用在集群中分配的執行器的個數。對于單個執行器（executor）所占用的資源，可以使用 –executor-memory 和 –executor-cores 來控制。Mesos 上還有一種動態共享 CPU 的方式。在這種模式下，每個 Spark 應用的內存占用仍然是固定且獨占的(仍由 spark.exexcutor.memory 決定)，但是如果該 Spark 應用沒有在某個機器上執行任務的話，那么其它應用可以占用該機器上的 CPU。這種模式對集群中有大量不是很活躍應用的場景非常有效，例如：集群中有很多不同用戶的 Spark shell session。但這種模式不適用于低延時的場景，因為當 Spark 應用需要使用 CPU 的時候，可能需要等待一段時間才能取得對 CPU 的使用權。要使用這種模式，只需要在 mesos://URL 上設置 spark.mesos.coarse 屬性為 false 即可。 值得注意的是，目前還沒有任何一種資源分配模式支持跨 Spark 應用的內存共享。如果你想通過這種方式共享數據，我們建議你可以單獨使用一個服務(例如：alluxio)，這樣就能實現多應用訪問同一個 RDD 的數據。 ## 動態資源分配 Spark 提供了一種基于負載來動態調節 Spark 應用資源占用的機制。這意味著，你的應用會在資源空閑的時間將其釋放給集群，需要時再重新申請。這一特性在多個應用 Spark 集群資源的情況下特別有用。 這個特性默認是禁止的，但是在所有的粗粒度集群管理器上都是可用的，如：i.e. 獨立部署模式[standalone mode](spark-standalone.html)，[YARN mode](running-on-yarn.html)，and 粗粒度模式[Mesos coarse-grained mode](running-on-mesos.html#mesos-run-modes)。 ### 配置和部署 要使用這一特性有兩個前提條件。首先，你的應用必須設置 spark.dynamicAllocation.enabled 為 true。其次，你必須在每個節點上啟動 external shuffle service，并將 spark.shuffle.service.enabled 設為 true。external shuffle service 的目的是在移除 executor 的時候，能夠保留 executor 輸出的 shuffle 文件(本文后續有更新的描述 [below](job-scheduling.html#graceful-decommission-of-executors))。啟用 external shuffle service 的方式在各個集群管理器上各不相同： 在 Spark 獨立部署的集群中，你只需要在 worker 啟動前設置 spark.shuffle.service.enabled 為 true 即可。 在 Mesos 粗粒度模式下，你需要在各個節點上運行 $SPARK_HOME/sbin/start-mesos-shuffle-service.sh 并設置 spark.shuffle.service.enabled為true即可。例如，你可以在Marathon來啟用這一功能。 在YARN模式下，需要按以下步驟在各個 NodeManager 上啟動： [here](running-on-yarn.html#configuring-the-external-shuffle-service). 所有其它的配置都是可選的，在 _spark.dynamicAllocation._ 和 _spark.shuffle.service._ 這兩個命名空間下有更加詳細的介紹 [configurations page](configuration.html#dynamic-allocation). ### 資源分配策略 總體上來說，Spark 應該在執行器空閑時將其關閉，而在后續要用時再申請。因為沒有一個固定的方法，可以預測一個執行器在后續是否馬上會被分配去執行任務，或者一個新分配的執行器實際上是空閑的，所以我們需要一個試探性的方法，來決定是否申請或是移除一個執行器。 #### 請求策略 一個啟用了動態分配的 Spark 應用會有等待任務需要調度的時候，申請額外的執行器。在這種情況下，必定意味著已有的執行器已經不足以同時執行所有未完成的任務。 Spark會分輪次來申請執行器。實際的資源申請，會在任務掛起 spark.dynamicAllocation.schedulerBacklogTimeout 秒后首次觸發，其后如果等待隊列中仍有掛起的任務，則每過 spark.dynamicAllocation.sustainedSchedulerBacklogTimeout 秒后觸發一次資源申請。另外，每一輪申請的執行器個數以指數形式增長。例如：一個 Spark 應用可能在首輪申請 1 個執行器，后續的輪次申請個數可能是 2 個、4 個、8 個......。 采用指數級增長策略的原因有兩個：第一，對于任何一個 Spark 應用如果只需要多申請少數幾個執行器的話，那么必須非常謹慎的啟動資源申請，這和 TCP 慢啟動有些類似；第二，如果一旦 Spark 應用確實需要申請多個執行器的話，那么可以確保其所需的計算資源及時增長。 #### 移除策略 移除執行器的策略就簡單得多了。Spark 應用會在某個執行器空閑超過 spark.dynamicAllocation.executorIdleTimeout 秒后將其刪除，在大多數情況下，執行器的移除條件和申請條件都是互斥的，也就是說，執行器在有等待執行任務掛起時，不應該空閑。 ### 優雅的關閉Executor(執行器) 非動態分配模式下，執行器可能的退出原因有執行失敗或是相關 Spark 應用已經退出。不管是哪種原因，執行器的所有狀態都已經不再需要，可以丟棄掉。但是在動態分配的情況下，執行器有可能在 Spark 應用運行期間被移除。這時候，如果 Spark 應用嘗試去訪問該執行器存儲的狀態，就必須重算這一部分數據。因此，Spark 需要一種機制，能夠優雅的關閉執行器，同時還保留其狀態數據。 這種需求對于混洗操作尤其重要。混洗過程中，Spark 執行器首先將 map 輸出寫到本地磁盤，同時執行器本身又是一個文件服務器，這樣其他執行器就能夠通過該執行器獲得對應的 map 結果數據。一旦有某些任務執行時間過長，動態分配有可能在混洗結束前移除任務異常的執行器，而這些被移除的執行器對應的數據將會被重新計算，但這些重算其實是不必要的。 要解決這一問題，就需要用到 external shuffle service，該服務在 Spark 1.2 引入。該服務在每個節點上都會啟動一個不依賴于任何 Spark 應用或執行器的獨立進程。一旦該服務啟用，Spark 執行器不再從各個執行器上獲取 shuffle 文件，轉而從這個 service 獲取。這意味著，任何執行器輸出的混洗狀態數據都可能存留時間比對應的執行器進程還長。 除了混洗文件之外，執行器也會在磁盤或者內存中緩存數。一旦執行器被移除，其緩存數據將無法訪問。這個問題目前還沒有解決。或許在未來的版本中，可能會采用外部混洗服務類似的方法，將緩存數據保存在堆外存儲中以解決這一問題。 # 應用內調度 在指定的 Spark 應用內部（對應同一 SparkContext 實例），多個線程可能并發地提交 Spark 作業（job）。在本節中，作業（job）是指，由 Spark action 算子（如：collect）觸發的一系列計算任務的集合。Spark 調度器是完全線程安全的，并且能夠支持 Spark 應用同時處理多個請求（比如：來自不同用戶的查詢）。 默認，Spark 應用內部使用 FIFO 調度策略。每個作業被劃分為多個階段（stage）（例如：map 階段和 reduce 階段），第一個作業在其啟動后會優先獲取所有的可用資源，然后是第二個作業再申請，再第三個……。如果前面的作業沒有把集群資源占滿，則后續的作業可以立即啟動運行，否則，后提交的作業會有明顯的延遲等待。 不過從 Spark 0.8 開始，Spark 也能支持各個作業間的公平（Fair）調度。公平調度時，Spark 以輪詢的方式給每個作業分配資源，因此所有的作業獲得的資源大體上是平均分配。這意味著，即使有大作業在運行，小的作業再提交也能立即獲得計算資源而不是等待前面的作業結束，大大減少了延遲時間。這種模式特別適合于多用戶配置。要啟用公平調度器，只需設置一下 SparkContext 中 spark.scheduler.mode 屬性為 FAIR 即可 : ``` val conf = new SparkConf().setMaster(...).setAppName(...) conf.set("spark.scheduler.mode", "FAIR") val sc = new SparkContext(conf) ``` ## 公平調度資源池 公平調度器還可以支持將作業分組放入資源池（pool），然后給每個資源池配置不同的選項（如：權重）。這樣你就可以給一些比較重要的作業創建一個“高優先級”資源池，或者你也可以把每個用戶的作業分到一組，這樣一來就是各個用戶平均分享集群資源，而不是各個作業平分集群資源。Spark 公平調度的實現方式基本都是模仿 [Hadoop Fair Scheduler](http://hadoop.apache.org/docs/current/hadoop-yarn/hadoop-yarn-site/FairScheduler.html)。來實現的。 默認情況下，新提交的作業都會進入到默認資源池中，不過作業對應于哪個資源池，可以在提交作業的線程中用 SparkContext.setLocalProperty 設定 spark.scheduler.pool 屬性。示例代碼如下 : ``` // Assuming sc is your SparkContext variable sc.setLocalProperty("spark.scheduler.pool", "pool1") ``` 一旦設好了局部屬性，所有該線程所提交的作業（即：在該線程中調用 action 算子，如：RDD.save/count/collect 等）都會使用這個資源池。這個設置是以線程為單位保存的，你很容易實現用同一線程來提交同一用戶的所有作業到同一個資源池中。同樣，如果需要清除資源池設置，只需在對應線程中調用如下代碼 : ``` sc.setLocalProperty("spark.scheduler.pool", null) ``` ## 資源池默認行為 默認地，各個資源池之間平分整個集群的資源（包括 default 資源池），但在資源池內部，默認情況下，作業是 FIFO 順序執行的。舉例來說，如果你為每個用戶創建了一個資源池，那么久意味著各個用戶之間共享整個集群的資源，但每個用戶自己提交的作業是按順序執行的，而不會出現后提交的作業搶占前面作業的資源。 ## 配置資源池屬性 資源池的屬性需要通過配置文件來指定。每個資源池都支持以下3個屬性 : * `schedulingMode`：可以是 FIFO 或 FAIR，控制資源池內部的作業是如何調度的。 * `weight`：控制資源池相對其他資源池，可以分配到資源的比例。默認所有資源池的 weight 都是 1。如果你將某個資源池的 weight 設為 2，那么該資源池中的資源將是其他池子的2倍。如果將 weight 設得很高，如 1000，可以實現資源池之間的調度優先級 – 也就是說，weight=1000 的資源池總能立即啟動其對應的作業。 * `minShare`：除了整體 weight 之外，每個資源池還能指定一個最小資源分配值（CPU 個數），管理員可能會需要這個設置。公平調度器總是會嘗試優先滿足所有活躍（active）資源池的最小資源分配值，然后再根據各個池子的 weight 來分配剩下的資源。因此，minShare 屬性能夠確保每個資源池都能至少獲得一定量的集群資源。minShare 的默認值是 0。 資源池屬性是一個 XML 文件，可以基于 conf/fairscheduler.xml.template 修改，然后在 [SparkConf](configuration.html#spark-properties)。的 spark.scheduler.allocation.file 屬性指定文件路徑： ``` conf.set("spark.scheduler.allocation.file", "/path/to/file") ``` 資源池 XML 配置文件格式如下，其中每個池子對應一個 &lt;pool&gt;元素，每個資源池可以有其獨立的配置 :&lt;/pool&gt; ``` <?xml version="1.0"?> <allocations> <pool name="production"> <schedulingMode>FAIR</schedulingMode> <weight>1</weight> <minShare>2</minShare> </pool> <pool name="test"> <schedulingMode>FIFO</schedulingMode> <weight>2</weight> <minShare>3</minShare> </pool> </allocations> ``` 完整的例子可以參考 conf/fairscheduler.xml.template。注意，沒有在配置文件中配置的資源池都會使用默認配置（schedulingMode : FIFO，weight : 1，minShare : 0）。