11.Spark之運行模式及原理

1.Spark運行模式

    Spark的運行模式多種多樣，靈活多變。

1. 部署在單機上時，既能夠用本地模式運行，也能夠用僞分佈式模式運行。

2. 以分佈式集羣方式部署時，也有衆多的運行模式可供選擇，這取決於集羣的實際狀況。底層的資源調度既能夠依賴於外部的資源調度框架（Mesos、Yarn），也可使用Spark內建的Standalone模式。

    實際應用中，Spark應用程序的運行模式取決於傳遞給SparkContext的MASTER環境變量的值，個別模式還須要依賴輔助的程序接口來配合使用，目前支持的MASTER環境變量由特定的字符串或URL所組成，以下所示：

• Local[N]：本地模式，使用N個線程。

• Local cluster[worker,core,Memory]：僞分佈式模式，能夠配置所須要啓動的虛擬工做節點的數量，以及每一個節點所管理的CPU數量和內存尺寸。

• Spark://hostname:port：Standalone模式，須要部署Spark到相關節點，URL爲Spark Master主機地址和端口。

• Mesos://hostname:port：Mesos模式，須要部署Spark和Mesos到相關節點，URL爲Mesos主機地址和端口。

• YARN standalone/YARN cluster：YARN模式一，主程序邏輯和任務都運行在YARN集羣中。

• YARN client：YARN模式二，主程序邏輯運行在本地，具體任務運行在YARN集羣中。

2.Spark基本工做流程

    運行模式儘管看起來差別很大，但整體來說都基於一個類似的工做流程。它們從根本上都是將Spark的應用分爲任務調度和任務執行兩個部分。下圖所示的是分佈式模式下，Spark的各個調度和執行模塊的大體框架圖。對於本地模式來講，其內部程序邏輯結構也是相似的，只是其中部分模塊有所簡化，例如集羣管理模塊簡化爲進程內部的線程池。

    

    具體來講，以SparkContext爲程序運行的總入口，在SparkContext的初始化過程當中，Spark會分別建立DAGScheduler做業調度和TaskScheduler任務調度兩級調度模塊。

    其中做業調度模塊是基於任務階段的高層調度模塊，它爲每一個Spark做業計算具備依賴關係的多個調度階段（一般根據shuffle來劃分），而後爲每一個階段構建出一組具體的任務（一般會考慮數據的本地性），而後以TaskSets（任務組）的形式提交給任務調度模塊來具體執行。而任務調度模塊則負責具體啓動任務、監控和彙報任務運行狀況。

    做業調度模塊和具體部署運行模式無關，在各類運行模式下邏輯相同。

    不一樣運行模式的區別主要體如今任務調度模塊。不一樣的部署和運行模式，根據底層資源調度方式的不一樣，各自實現了本身特定的任務調度模塊，用來將任務實際調度給對應的計算資源。

3.相關基本類

• TaskScheduler / SchedulerBackend
  

    爲了抽象出一個公共的接口供DAGScheduler做業調度模塊使用，全部的這些運行模式實現的任務調度模塊都是基於這兩個接口（Trait）的：TaskScheduler及SchedulerBackend。

    TaskSceduler的實現主要用於與DAGScheduler交互，負責任務的具體調度和運行，其核心接口是submitTasks和cancleTasks。

    SchedulerBackend的實現是與底層資源調度系統交互（如Mesos/YARN），配合TaskSceduler實現具體任務執行所須要的資源分配，核心接口是receiveOffers。

• TaskSchedulerImpl

    TaskSchedulerImpl實現了TaskSceduler接口，提供了大多數本地和分佈式運行調度模式的任務調度接口。

    此外它還實現了resourceOffers和statusUpdate這兩個接口供Backend調用，用於提供調度資源和更新任務狀態。

    另外，在提交任務和更新狀態等階段，TaskSchedulerImpl都會調用Backend的receiveOffers函數，用於發起一次任務資源調度請求。

• Executor

    實際任務的運行，最終都由Executor類來執行，Executor對每個任務建立一個TaskRunner類，交給線程池運行。運行結果經過ExecutorBackend接口返回。

4.Local模式

• 部署及程序運行

  Local模式，顧名思義就是在本地運行。若是不加任何配置，Spark默認設置爲Local模式。以SparkPi爲例，Local模式下的應用程序的啓動命令以下：

         ./bin/run-example org.apache.spark.examples.SparkPi local

  在SparkPi的具體實現中，是根據傳入的參數來選擇運行模式的（如上例中的local）。也能夠在代碼中配置Master來指定運行模式，但爲了可以使應用程序在各類部署環境下使用，不建議把與運行環境相關的設置在代碼中寫死。

• 內部實現原理

  Local本地模式使用LocalBackend配合TaskSchedulerImpl，內部邏輯結構如圖所示：

  LocalBackend響應Scheduler的receiveOffers請求，根據可用的CPU核的設定值[N]直接生成CPU資源返回給Scheduler，並經過Executor類在線程池中依次啓動和運行Scheduler返回的任務列表，其核心事件循環由內部類LocalActor以Akka Actor的消息處理形式來實現。

  由於Local模式無需配置，同時全部的代碼都在本地進程中執行，因此經常用來做爲快速驗證代碼和跟蹤調試的手段。

5.Standalone模式

• 部署及程序運行

  在該模式中，Spark集羣由Master、Worker節點構成，程序經過與Master節點交互申請所需的資源，Worker節點負責具體Executor的啓動運行。部署也很簡單，只須要將編譯好或者下載的Spark發佈版本複製到各個集羣節點便可。爲了方便經過啓動腳本使用，部署的路徑最好一致。

  手動啓動集羣

  在做爲Master節點的服務器上經過命令手動啓動集羣：

         ./sbin/start-master.sh

  接下來在打算做爲Worker節點的服務器上啓動一個或多個Worker：

         ./bin/spark-class org.apache.spark.deploy.worker.Worker spark://MasterURL:PORT

  注：spark://MasterURL:PORT中的MasterURL和PORT能夠在./sbin/start-master.sh輸出信息中找到。

  腳本啓動集羣

  首先在Spark的Conf目錄下建立一個名爲slaves的文件，內容包含全部做爲Worker節點的主機名（或IP），按行分隔。同時，Master節點必須能經過SSH祕鑰登陸的形式訪問Worker節點（不須要手動輸入密碼）。配置好以後，就能夠經過如下腳本啓動、中止Standalone模式的集羣了。

  sbin/start-master.sh：在執行該腳本的機器上啓動一個Spark Master實例。

  sbin/start-slaves.sh：在slaves文件中指定的每個節點上啓動一個或多個Spark Worker實例。

  sbin/start-all.sh：執行上述兩步。

  sbin/stop-master.sh：關閉由sbin/start-master.sh啓動的Master實例。

  sbin/stop-slaves.sh：關閉由sbin/start-slaves.sh啓動的Worker實例。

  sbin/stop-all.sh：執行上述兩步。

  集羣參數配置

  經過conf/spark-env.sh配置環境變量，你能夠進一步控制Spark集羣的運行參數。下面列出一些主要可供配置的參數。

  SPARK_MASTER_IP：Master節點的IP地址。

  SPARK_MASTER_PORT：Spark Master的工做端口（默認爲7077）。

  SPARK_MASTER_WEBUI_PORT：Master節點監控網頁的端口（默認爲8080）。

  SPARK_WORKER_PORT：Spark Worker的工做端口（默認爲隨機）。

  SPARK_WORKER_DIR：Spark應用程序，包括Log記錄等使用的目錄（默認爲SPARK_HOME/work）。

  SPARK_WORKER_CORES：Spark Worker管理的CPU核的數量（默認爲使用全部核）。

  SPARK_WORKER_MEMORY：Spark Worker管理的內存數量（默認爲總內存減1GB）。

  SPARK_WORKER_WEBUI_PORT：Spark Worker的網頁端口（默認爲8081）。

  SPARK_WORKER_INSTANCES：每一個物理節點所運行的Worker進程的數量（默認爲1），經過配置這個參數能夠在一臺機器上運行多個Spark Worker進程，但須要注意的是須要對應地調整每一個Worker所管理的CPU核數和內存的數量。

  SPARK_DAEMON_MEMORY：Spark Master和Spark Worker本身所使用內存的大小（默認爲512MB）。

  SPARK_DAEMON_JAVA_OPTS：傳遞給Spark Master和Spark Worker的JVM參數。

• 內部實現原理

  該模式使用SparkDeploySchedulerBackend配合TaskSchedulerImpl工做，而SparkDeploySchedulerBackend自己拓展自CoarseGrainedSchedulerBackend。

  
  

  CoarseGrainedSchedulerBackend是一個基於Akka Actor實現的粗粒度的資源調度類，在整個Spark做業運行期間，CoarseGrainedSchedulerBackend會監聽並持有註冊給它的Executor資源（相對於細粒度的調度，Executor基於每一個任務的生命週期建立和銷燬），而且在接受Executor註冊、狀態更新、響應Scheduler請求等各類時刻，根據現有Executor資源發起任務調度流程。

  Executor自己是能夠經過各類途徑啓動的，對應的在Spark Standalone模式中，SparkDeploySchedulerBackend經過Client類向Spark Master發送請求，在獨立部署的Spark集羣中啓動CoarseGrainedExecutorBackend，根據所需的CPU資源的數量，一個或多個CoarseGrainedExecutorBackend在Spark Worker節點上啓動並註冊給SparkDeploySchedulerBackend的Driver Actor。

  完成所需Actor的啓動後，任務調度就在SparkDeploySchedulerBackend和CoarseGrainedExecutorBackend的Actor之間直接完成。

6.Local cluster模式

• 部署及程序運行

  Local cluster僞分佈式模式，實際是在SparkContext初始化的過程當中，在本地啓動一個全部服務都在單機上運行的僞分佈Spark集羣，因此從部署的角度來講無須作任何準備工做。以SparkPi爲例，僞分佈式模式下的應用程序的啓動命令的示例以下：

         ./bin/run-example org.apache.spark.examples.SparkPi local-cluster[2,2,1024]

  上面的僞分佈式模式啓動兩個Worker，每一個Worker管理兩個CPU和1024MB的內存。

• 內部實現原理

  僞分佈式模式是基於Standalone模式來實現的，除了啓動Master和Worker的位置不一樣，在集羣啓動完畢後，後續的應用程序的工做流程及資源的調度流程與Standalone模式徹底相同。

7.Mesos模式

• 部署及程序運行

  首先須要部署Mesos到全部的工做節點上，請參考http://mesos.apache.org/gettingstarted。

  以後須要建立一個Spark發佈包，能夠經過make-distribution.sh腳原本建立。這個腳本將Spark運行所需的配置文件和依賴關係複製到dist目錄中，再將這個目錄打包上傳到HDFS中共Mesos調用。

  此外，還需在spark-env.sh文件中配置如下幾個環境變量，須要注意的是若是你的Mesos集羣使用Zookeeper來管理Master節點，那Master對應須要配置的是Zookeeper的主機和端口。

  MESOS_NATIVE_LIBRARY：libmesos.so的路徑。

  SPARK_EXECUTOR_URI：Spark發佈包的地址。

  MASTER：格式爲mesos://HOST:PORT，Mesos Master的主機和端口（Mesos端口默認爲5050）

  以SparkPi爲例，在Mesos模式下的應用程序啓動命令以下：

         ./bin/run-example org.apache.spark.examples.SparkPi mesos://10.0.2.31:5050

• 內部實現原理

  在該模式中，資源的調度能夠分爲粗粒度和細粒度調度兩種，相應的Spark分別使用CoarseMesosSchedulerBackend和MesosSchedulerBackend來配合TaskSchedulerImpl工做。

8.YARN standalone/YARN cluster模式

• 部署及程序運行

  0.9以前使用YARN standalone，1.0之後改成YARN cluster，經過Hadoop YARN框架來調度Spark應用所須要的資源。要將Spark集羣運行在YARN模式下，首先部署一個YARN集羣供該模式使用。實際中，這二者的因果關係每每是反過來的，這是由於用戶有了一個YARN集羣，也但願將Spark歸入YARN的調度管理之下，這樣有利於系統的資源共享。

  在該模式下，須要經過額外的輔助程序來啓動應用。
  

• 內部實現原理

  相對於其餘模式，YARN cluster模式須要由外部程序輔助啓動APP。用戶的應用程序經過輔助的YARN Client類啓動。

  Client類經過YARN Client API提交請求，在Hadoop集羣上啓動一個Spark ApplicationMaster，它首先註冊本身爲一個YARN ApplicationMaster，以後啓動用戶程序，SparkContext在用戶程序中初始化時，使用CoarseGrainedSchedulerBackend配合YARNClusterScheduler（只是對TaskScheduler的一個簡單封裝，增長了對Executor的等待邏輯等）。

  根據Client類傳遞的參數，Spark ApplicationMaster經過YARN ResourceManager/NodeManager的接口在集羣中啓動若干個Container（容器），用於運行CoarseGrainedExecutorBackend。CoarseGrainedExecutorBackend在啓動過程當中會向CoarseGrainedSchedulerBackend註冊。以後的任務調度流程同其餘Cluster模式。

9.YARN client模式

• 部署及程序運行

  YARN client模式的部署準備工做和YARN standalone模式徹底一致。因爲該模式不直接暴露前面提到的用於提交YARN程序的輔助程序，因此許多參數是經過環境變量來設置的，能夠配置的環境變量以下。

  SPARK_JAR：Spark自身JAR包的路徑。

  SPARK_YARN_APP_JAR：應用APP的JAR包路徑，在1.0版本中已經移除。

  SPARK_WORKER_INSTANCES：須要啓動的Worker的數量（默認爲2）。

  SPARK_WORKER_CORES：每一個Worker使用的CPU Core的數量（默認爲1）。

  SPARK_WORKER_MEMORY：每一個Worker使用的內存數量（默認爲1GB）。

  SPARK_MASTER_MEMORY：Master節點的內存大小（默認爲512MB）。

  SPARK_YARN_APP_NAME：應用程序在YARN框架中使用的名字（默認爲Spark）。

  SPARK_YARN_QUEUE：資源分配時使用的YARN的隊列（默認爲'default'）。

  SPARK_YARN_DIST_FILES：須要隨做業分發到集羣上的文件。

  SPARK_YARN_DIST_ARCHIVES：須要隨做業分發到集羣上的壓縮包。

  這些環境變量能夠在spark-env.sh中配置，也能夠在啓動應用時設置。

• 內部實現原理

  在YARN cluster模式中，應用程序（包括SparkContext）都是做爲YARN框架所須要的Application Master，再由YARN ResourceManager爲其分配的一個隨機節點上運行。而在YARN client模式中，SparkContext運行在本地，該模式適用於應用程序自己須要在本地進行交互的場合，如Spark Shell、Shark等。

  SparkContext在初始化過程當中啓動YARNClientSchedulerBackend（拓展自CoarseGrainedSchedulerBackend），該Backend進一步調用org.apache.spark.deploy.yarn.Client在遠程啓動一個WorkerLauncher做爲Spark的Application Master，相比於YARN standalone模式，WorkerLauncher再也不負責用戶程序的啓動（已在客戶端本地啓動），而只是啓動Container運行CoarseGrainedExecutorBackend與客戶端本地的Driver進行通訊，後續任務調度流程相同。