SparkContext 初始化內部原理

劉彬同學準備寫一系列spark實戰系列，本文是第一篇，SparkContext初始化內部原理！贊！推薦給大家，希望大家喜歡和支持！

如果編寫Spark程序，那麼第??代碼就是new SparkContext().setMaster(「」).setAppName(「xx」),可以說SparkContext是整個Spark 計算的啟動器，只有將sparkContext 啟動起來，後續的關於調 度，存儲，計算相關的操作才能夠運?，本?基於spark2.x源碼概述關於SparkContext??所包含的啟動項都有哪些以及這些啟動項的作?是什麼，之後在說?下關於SparkEnv環境創建的過程。

閱讀本?最好打開spark源碼參考著看，可以在git上?打開spark關於sparkContext的代碼，地址為：https://github.com/apache/spark/blob/c5f9b89dda40ffaa4622a7ba2b3d0605dbe815c0/core/src/main/scala/org/apache/spark/SparkContext.scala#L73

01

SparkContext內部組件：

如圖為SparkContext內部的?些組件：

• SparkEnv ：Spark運?時環境，Spark 中任務執?是通過Executor，所有的Executor都有??的執?環境SparkEnv，在Driver中也包含了SparkEnv，為了保證Local模式的運?，SparkEnv內部還提 供了不同的組件，來實現不同的功能

• LiveListenerBus：SparkContext中的事件匯流排，可以接收各個使?者的事件，非同步將SparkListenerevent傳遞給註冊的SparkListener

• Spark UI ：Spark的?戶界?，SparkUI間接依賴於計算引擎，調度引擎，存儲引擎，Job，Stage，Executor等組件的監控都會以SparkListenerEvent的形式傳遞給LiveListenerBus，SparkUI將從各 個SparkListener中讀取數據並顯?在web界?

• SparkStatusTracker：?於監控作業和Stage進度狀態的低級API

• ConsoleProgressBar ：定期從sc.statusTracker獲得active stage的狀態信息，展?到進度條［在SparkUI上?可以看到進度條］，會有?定的延時。內部有?個timer 500ms refresh?遍

• DAGScheduler：DAG調度器，是Spark調度系統中重要的組件之?，負責創建Job，將DAG的RDD劃分到不同的Stage，提交stage等，SparkUI中有關Job和Stage監控數據都來?DAGScheduer

• TaskScheduler：Task調度器，是Spark調度系統中重要的組件之?，負責將任務發送到集群，運?，如果有失敗的任務則重新執?，之後返回給DAGScheduler，TaskScheduler調度的Task是由 DAGScheduler創建的，所以DAGScheduler是TaskScheduler前置調度。

• HeatbeatReceiver：?跳接收器，所有的Executor都會向HeatbeatReceiver發送?跳信息，HeatbeatReceiver接收到?跳之後，先更新Executor最後可?時間，然後將此信息交給TaskScheduler。

• ContextCleaner：非同步清理RDD、shuffle和?播狀態信息

• EventLoggingListener：將事件持久化到存儲的監聽器，是SparkContext的可選組件，當spark.eventLog.enable

• ExecutorAllocationManager: Executor動態分配管理器，根據?作負載動態調整Executor數量，當在配置spark.dynamicAlloction.enabled屬性為true的情況下，在?local模式下或者 spark.dynamicAllcation.testing屬性為true時啟?

• ShutdownHookManager：設置關閉鉤?的管理器，可以給應?設置鉤?，這樣當JVM退出的時候就會執?清理?作

除了以上這些SparkContext包含的內部組件，還包括如下?些屬性：

• creationSite：CallSite類型，保存著線程棧中最靠近棧頂的?戶定義的類和最靠近棧底的Scala或者Spark核?類的信息，其中ShortForm屬性保存著上述信息的間斷描述，LongForm屬性保存著上述 信息的完整描述，具體的信息可以參閱源碼部分地址為：core/src/main/scala/org/apache/spark/util/Utils.scala/getCallSite

• allowMulitipleContext : 是否允許多個SparkContext實例，默認為False，可以通過設置Spark.Driver.allowMulitipleContexts來控制

• startTime：標記sparkContext的啟動時間戳

• stopped：標記sparkContext是否處於停?狀態，采?原?類型AtomicBoolean

• addedFiles：?於每個本地?件的URL與添加此?件到到addedFiles時的時間戳之間的映射緩存 new ConcurrentHashMap[String, Long]

• addedJars:?於每個本地Jar?件的URL與添加此?件到addedJars時的時間戳之間的映射緩存 new ConcurrentHashMap[String, Long]

• persistentRdds:?於對所有持久化的RDD保持跟蹤

• executorEnvs：?於存儲環境變數，將?於Executor執?的時候使?

• sparkUser：當前系統的登錄?戶，可以通過環境變數SPARK_USER來設置 通過Utils.getCurrentUserName()獲取

• checkpointDir：RDD計算過程中?於記錄RDD檢查點的?錄

• localProperties：InheritableThreadLocal保護的線程，其中的屬性值可以沿著線程棧?直傳遞下去

• _conf:SparkContext的配置，會先調?config的clone?法，在進?驗證配置，是否設置了spark.master和spark.app.name

• jars:?戶提交的jar?件，當選擇部署模式為yarn時，

_eventLogDir:事件?志的路徑,當spark.enabled屬性為true時啟?，默認為/tmp/spark-events,也可以通過spark.eventLog.dir來指定?錄 _eventLogCoder:事件?志的壓縮演算法，當spark.eventLog.enabled屬性與spark.eventLog.compress屬性為true時，壓縮演算法默認為lz4，也可以通過spark.io.compression.codec屬性指定，?前?持lzf,snappy和lz4

• _hadoopConfiguration:Hadoop配置信息，如果系統屬性SPARK_YARN_MODE為true或者環境變數SPARK_YARN_MODEL為true，那麼將會是YARN的配置，否則為Hadoop的配置

• _executorMemtory:Executor內存??，默認為1024MB，可以通過設置環境變數（SPARK_MEM或者SPARK_EXECUTOR_MEMORY）或者Spark.executor.memory屬性指定其中Spark.executor.memory優先順序最?

• _applicationId:當前應?的標識，TaskScheduler啟動後會創建應?標識，通過調取TaskScheduler的ApplicationId獲取的

• _applicationAttempId:當前應?嘗試執?的標識，SparkDriver在執?時會多次嘗試，每次嘗試都會?成?個標識來代表應?嘗試執?的?份

• _listenerBusStarted:LiveListenerBus是否已經啟動的標記

• nextShuffleId:類型為AtomicInteger，?於?成下?個shuffle標識

• nextRddId：類型為atomicInteger，?於?成下?個rdd標識

02

初始化具體流程

以上可以在github上打開spark源碼進?邊看?章邊看源碼，你會受益良多。 在這?推薦?個github源碼閱讀插件Insight.io for Github 在chrome擴展程序里可以直接查詢。

加入技術討論群

《大數據和雲計算技術》社區人數已經2500+，歡迎大家加下面助手微信，拉大家進群，自由交流。

喜歡釘釘群的，可以掃描下面二維碼：

喜歡QQ群的，可以掃描下面二維碼：

歡迎大家通過二維碼打賞支持技術社區（英雄請留名，社區感謝您，打賞次數超過55+）：

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！