Spark internal - 多樣化的運行模式 （下）

上一篇中介紹了Spark的各類運行模式的基本流程和相關實現，這裏主要分析一下各類運行模式中涉及到的一些細節問題的流程和實現

 

Spark的各類運行模式雖然啓動方式，運行位置，調度手段有所不一樣，但它們所要完成的任務基本都是一致的，就是在合適的位置安全可靠的根據用戶的配置和Job的須要管理和運行Task，這裏粗略的列舉一下在運行調度過程當中各類須要考慮的問題

 

• 環境變量的傳遞

• Jar包和各類依賴文件的分發

• Task的管理和序列化等

• 用戶參數配置

• 用戶及權限控制

 

 

環境變量的傳遞

 

Spark的運行參數有很大一部分是經過環境變量來設置的，例如Executor的內存設置，Library路徑等等。Local模式固然不存在環境變量的傳遞問題，在Cluster模式下，就須要將環境變量傳遞到遠端JVM環境中去

 

SparkContext在初始化過程當中 須要傳遞給Executor的環境變量，會在executorEnvs變量中（HashMap）中收集起來

 

而具體如何將這些變量設置到Executor的環境中，取決於Executor的Launch方式

 

在Spark Standalone模式中，這些變量被封裝在org.apache.spark.deploy.Command中，交給AppClient啓動遠程Executor，Command經由Spark  Master經過Actor再次轉發給合適的Worker，Worker經過ExecutorRunner構建Java.lang.Process運行ExecutorBackend，環境變量在ExecutorRunner中傳遞給java.lang.ProcessBuilder.environment完成整個傳遞過程

 

在Mesos相關模式中，這些環境變量被設置到org.apache.mesos.Protos.Environment中，在經過MesosLaunch  Task時交給Mesos完成分發工做

 

在yarn-standalone模式中，這些環境變量首先要經過Yarn  Client 設置到Spark AM的運行環境中，基本就是Client類運行環境中以SPARK開頭的環境變量所有設置到ContainerLaunchContext中，AM經過WorkerRunnable進一步將它們設置到運行Executor所用的ContainerLaunchContext中

 

Yarn-client模式與yarn-standalone模式大體相同，雖然SparkContext運行在本地，executor所需的環境變量仍是經過ContainerLaunchContext經AM中轉發給Executor

 

能夠注意到，在Yarn相關模式中，並無使用到SparkContext收集的executorEnvs，主要是由於Yarn  Standalone模式下Sparkcontext自己就是在遠程運行的，所以在Yarn Client中單獨實現了相關代碼

 

Jar包和各類依賴文件的分發

 

Spark程序的運行依賴大體分兩類, 一是Spark runtime及其依賴，二是應用程序自身的額外依賴

對於Local模式而言，不存在Jar包分發的問題

對於第一類依賴

 

在Spark Standalone模式中，整個環境隨Spark部署到各個節點中，所以也不存在runtime  Jar包分發的問題

 

Mesos相關模式下，Mesos自己須要部署到各個節點，SparkRuntime能夠和Standalone模式同樣部署到各個節點中，也能夠上傳到Mesos能夠讀取的地方好比HDFS上，而後經過配置spark.executor.uri通知Mesos相關的SchedulerBackend，它們會將該URL傳遞給Mesos，Mesos在Launch任務時會從指定位置獲取相關文件

 

而Spark 應用程序所額外依賴的文件，在上述模式中能夠經過參數將URL傳遞給SparkContext，對於本地文件SparkContext將啓動一個HttpServer用於其它節點讀取相關文件，其它如HDFS和外部HTTP等地址上的文件則原封不動，而後這些額外依賴文件的URL在TaskSetmanager中和Task自己一塊兒被序列化後發送給Executor，Executor再反序列化獲得URL並傳遞給ExecutorURLClassLoader使用

 

在Yarn相關模式中，Runtime和程序運行所依賴的文件首先經過HDFS  Client API上傳到Job的.sparkStaging目錄下，而後將對應的文件和URL映射關係經過containerLaunchContext.setLocalResources函數通知Yarn，Yarn的NodeManager在Launch  container的時候會從指定URL處下載相關文件做爲運行環境的一部分。上面的步驟對於Spark  AM來講是充分的，而對於須要進一步分發到Executor的運行環境中的文件來講，AM還須要在建立Executor的Container的時候一樣調用setLocalResources函數，AM是如何得到對應的文件和URL列表的呢，其實就是SparkYarn  Client將這些文件的相關屬性如URL，時間戳，尺寸等信息打包成字符串，經過特定的環境變量（SPARK_YARN_CACHE_XXX ）傳遞給AM，AM再把它們從環境變量中還原成所需文件列表

 

 

Task管理和序列化

 

Task的運行要解決的問題不外乎就是如何以正確的順序，有效地管理和分派任務，如何將Task及運行所需相關數據有效地發送到遠端，以及收集運行結果

 

Task的派發源起於DAGScheduler調用TaskScheduler.submitTasks將一個Stage相關的一組Task一塊兒提交調度。

 

在TaskSchedulerImpl中，這一組Task被交給一個新的TaskSetManager實例進行管理，全部的TaskSetManager經由SchedulableBuilder根據特定的調度策略進行排序，在TaskSchedulerImpl的resourceOffers函數中，當前被選擇的TaskSetManager的ResourceOffer函數被調用並返回包含了序列化任務數據的TaskDescription，最後這些TaskDescription再由SchedulerBackend派發到ExecutorBackend去執行

 

系列化的過程當中，上一節中所述App依賴文件相關屬性URL等經過DataOutPutStream寫出，而Task自己經過可配置的Serializer來序列化，當前可配製的Serializer包括如JavaSerializer ，KryoSerializer等

 

Task的運行結果在Executor端被序列化併發送回SchedulerBackend，因爲受到Akka  Frame Size尺寸的限制，若是運行結果數據過大，結果會存儲到BlockManager中，這時候發送到SchedulerBackend的是對應數據的BlockID，TaskScheduler最終會調用TaskResultGetter在線程池中以異步的方式讀取結果，TaskSetManager再根據運行結果更新任務狀態（好比失敗重試等）並彙報給DAGScheduler等

 

 

用戶參數配置

 

Spark的用戶參數配置途徑不少，除了環境變量之外，能夠經過Spark.conf文件設置，也能夠經過修改系統屬性設置  "spark.*"

 

而這些配置參數的使用環境也不少樣化，有些在Sparkcontext本地使用（除了yarn-standalone模式），有些須要分發到Cluster集羣中去

 

在SparkContext中解析和使用，好比spark.master，spark.app.names,  spark.jars等等，一般用於配置SparkContext運行參數，建立Executor啓動環境等

 

發送給Executor的參數又分兩部分

 

一部分在ExecutorBackend初始化過程當中須要使用的系統變量，會經過SparkContext在初始化過程當中讀取並設置到環境變量中去，在經過前面所述的方式，使用對應的底層資源調度系統設置到運行容器的環境變量中

 

另外一部分在Executor中才使用的以"spark.*"開頭的參數，則經過ExecutorBackend向SchedulerBackend的註冊過程，在註冊確認函數中傳遞給ExecutorBackend再在Executor的初始化過程當中設置到SparkConf中

 

整體看來，這些參數配置的方式和分發途徑有些不太統一，稍顯混亂，大概還有改進的餘地

 

用戶及權限控制

 

Spark的Task在Executor中運行時，使用hadoop的UerGroupInfomation.doAs 函數將整個Task的運行環境包裝起來以特定的sparkUser的身份運行。這樣作的目的主要是使得Spark的task在與Hadoop交互時，使用特定的用戶而不是Executor啓動時所用的用戶身份，這有利於在集羣中區分Spark  Cluster的運行用戶和實際使用集羣的APP用戶身份，以及HDFS等權限控制

 

用戶名在Executor中經過SPARK_USER環境變量獲取

 

對於Local模式來講，SPARK_USER環境變量就是當前JVM環境下設定的值，固然對Local模式來講實際上也是不須要doAs的，Executor中若是SPARK_USER變量未設定或者與當前用戶名一致，會跳過doAs直接執行task  launch相關函數

 

傳遞用戶身份的問題容易解決，比較麻煩的是身份的認證，例如將Spark運行在經過Kerberos管理權限的Hadoop集羣中，這須要完成客戶端的身份認證，Security 相關祕鑰或Token的獲取，分發，更新，失效等工做，在保證效率的同時，還要確保整個過程的安全性，目前的Spark代碼對這一方面尚未完善的實現方案，可是有一些提案和Patch正在進行中。

轉自：http://blog.csdn.net/colorant/article/details/18603965