即日起開始spark源碼閱讀之旅，這個過程是相當痛苦的，也許有大量的看不懂，但是每天一個方法，一點點看，相信總歸會有極大地提高的。那么下面開始：
創建sparkConf對象，那么究竟它干了什么了類,從代碼層面，我們可以看到我們需要setMaster啊，setAppName啊，set blabla啊。。。等等~
val sparkConf = new SparkConf().setMaster("local").setAppName("TopActiveLocations").set("spark.executor.memory", "3g")
那么我們就一點一點看一下,SparkConf是怎么實現的：
class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging { import SparkConf._ /** Create a SparkConf that loads defaults from system properties and the classpath */ def this() = this(true) private val settings = new ConcurrentHashMap[String, String]() if (loadDefaults) { // Load any spark.* system properties for ((key, value) <- Utils.getSystemProperties if key.startsWith("spark.")) { set(key, value) } }
/** Set a configuration variable. */
def set(key: String, value: String): SparkConf = {
if (key == null) {
throw new NullPointerException("null key")
}
if (value == null) {
throw new NullPointerException("null value for " + key)
}
logDeprecationWarning(key)
settings.put(key, value)
this
}
你會發現，它聲明了一個settings的ConcurrentHashMap，用的正是 java.util.concurrent.ConcurrentHashMap，從ConcurrentHashMap代碼中可以看出，它引入了一個“分段鎖”的概念，具體可以理解為把一個大的Map拆分成N個小的HashTable，根據key.hashCode()來決定把key放到哪個HashTable中。。。。額。。。我們是在玩spark,言歸正傳。
然后呢在聲明對象是，SparkConf傳入的是一個boolean類型的變量，這個變量的作用是是否加載Spark的conf下的配置信息，這個從def this() = this(true)可以看出，默認是為true的，這也就是為什么我們代碼中提交集群，不用去專門set配置項的原因，而且大部分伙計不知道這里還可以傳值~
隨后，如果為true的情況下，它會去getSystemProperties進行加載。
def getSystemProperties: Map[String, String] = {
System.getProperties.stringPropertyNames().asScala
.map(key => (key, System.getProperty(key))).toMap
}
/** * Enumerates all key/value pairs in the specified hashtable * and omits the property if the key or value is not a string. * @param h the hashtable */ private synchronized void enumerateStringProperties(Hashtable<String, String> h) { if (defaults != null) { defaults.enumerateStringProperties(h); } for (Enumeration e = keys() ; e.hasMoreElements() ;) { Object k = e.nextElement(); Object v = get(k); if (k instanceof String && v instanceof String) { h.put((String) k, (String) v); } } }
最終都存入了之前的map中，我們繼續深入創建sparkContext對象。
val sc = new SparkContext(sparkConf)
然后我發現 它干了一大堆一大堆的變態的事情,首先我們看一下sparkContext的構造器：
class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationClient { // The call site where this SparkContext was constructed. private val creationSite: CallSite = Utils.getCallSite() // If true, log warnings instead of throwing exceptions when multiple SparkContexts are active private val allowMultipleContexts: Boolean = config.getBoolean("spark.driver.allowMultipleContexts", false) // In order to prevent multiple SparkContexts from being active at the same time, mark this // context as having started construction. // NOTE: this must be placed at the beginning of the SparkContext constructor. SparkContext.markPartiallyConstructed(this, allowMultipleContexts) val startTime = System.currentTimeMillis()
首先，創建了CallSite對象，那么這個對象是干什么的呢，它存儲了線程棧中最靠近棧頂的用戶類及最靠近棧底的Scala或者Spark核心類信息。
這里，config.getBoolean("spark.driver.allowMultipleContexts", false)默認為false，曾經我以為只能在spark中創建一個Sparkcontext對象，其實可以創建多個（我勒個去啊，那是不是說明可以同時創建streaming對象以及sparkContext對象，將streaming與sparksql同時聲明，一起做數據處理了，有待驗證） 如果需要創建多個，就在配置參數中設置為true. markPartiallyConstructed會確保其唯一性。
接下來呢會拷貝config，並且進行默認值賦值，與為空判斷，這里可以看到spark.master 和spark.app.name 是必須設置的，否則會拋出。
隨之調用
SparkEnv.createDriverEnv(conf, isLocal, listenerBus, SparkContext.numDriverCores(master))方法，創建SparkEnv.查閱資料，SparkEnv呢，又干了N多事情如下：
1.創建安全管理器SecurityManager；
SecurityManager主要對權限、賬號進行設置，如果使用Hadoop YARN作為集群管理器，則需要使用證書生成secret key登錄，最后給當前系統設置默認的口令認證實例。
2.基於Akka的分布式消息系統ActorSystem
Scala認為Java線程通過共享數據以及通過鎖來維護共享數據的一致性是糟糕的做法，容易引起鎖的爭用，降低並發程序的性能，甚至會引入死鎖的問題。在Scala中只需要自定義類型繼承Actor，並且提供act方法，就如同Java里實現Runnable接口，需要實現run方法一樣。但是不能直接調用act方法，而是通過發送消息的方式（Scala發送消息是異步的）傳遞數據。
3.下來呢，該創建MapOutputTrackerMaster或MapOutputTrackerWorker，那么他倆是什么呢?map任務的狀態正是由Executor向持有的MapOutputTracker-MasterActor發送消息，將map任務狀態同步到mapOutputTracker的mapStatuses，Executor究竟是如何找到MapOutputTrackerMasterActor的？registerOrLookup方法通過調用AkkaUtils.makeDriverRef找到MapOutputTrackerMasterActor，利用ActorSystem提供的分布式消息機制實現的.
4.隨之開始對ShuffleManager實例進行創建及加載。
ShuffleManager默認為通過反射方式生成的SortShuffleManager的實例，可以修改屬性spark.shuffle.manager為hash來顯式控制使用HashShuffleManager。這里再說明下，什么是shuffle?shuffle就是個混洗的過程,同一個作業會被划分為多個任務在多個節點上並行執行，reduce的輸入可能存在於多個節點上，需要通過“洗牌”將所有reduce的輸入匯總起來，這個過程就是shuffle。 那么spark是通過反射，來加載對應配置項的實體類:
ShuffleMemoryManager負責管理shuffle線程占有內存的分配與釋放，並通過thread-Memory：mutable.HashMap[Long，Long]緩存每個線程的內存字節數。出，shuffle所有線程占用的最大內存的計算公式為：
Java運行時最大內存*Spark的shuffle最大內存占比*Spark的安全內存占比，可以配置屬性spark.shuffle.memoryFraction修改Spark的shuffle最大內存占比，配置屬性spark.shuffle.safetyFraction修改Spark的安全內存，如下代碼：
5.下來，創建BlockManager，BlockManager負責對Block的管理，只有在BlockManager的初始化方法initialize被調用后，它才是有效的。BlockManager作為存儲系統的一部分。這么就繼續深入，圍繞BlockManager進行閱讀。
查閱資料，BlockManager主要由以下部分組成：
·shuffle客戶端ShuffleClient；
·BlockManagerMaster（對存在於所有Executor上的BlockManager統一管理）；
·磁盤塊管理器DiskBlockManager
·磁盤存儲DiskStore；
·Tachyon存儲TachyonStore；
·非廣播Block清理器metadataCleaner和廣播Block清理器broadcastCleaner；
·壓縮算法實現
ShuffleServerId默認使用當前BlockManager的BlockManagerId。BlockManager的初始化：
那么BlockManager的實質運行機制如下圖：
（1）表示Executor 的BlockManager中的BlockManagerMaster與Driver的BlockManagerActor進行消息通信，比如注冊BlockManager、更新Block的信息、獲取Block所在的BlockManager、刪除Executor等。
（2）是shuffleRead與shufflewrite過程，也是BlockManager的讀寫操作。
（3）當內存不足時，寫入磁盤，寫入磁盤的數據也是由DiskBlockManager進行管理。
（4）通過訪問遠端節點的Executor的BlockManager中的TransportServer提供的RPC服務下載或者上傳Block；
（5）遠端節點的Executor的BlockManager訪問本地Executor的BlockManager中的TransportServer提供的RPC服務下載或者上傳Block；
（6）當存儲體系選擇Tachyon作為存儲時，對於BlockManager的讀寫操作實際調用了TachyonStore的putBytes、putArray、putIterator、getBytes、getValues等。
以上過程就發生在我們提交jar包或啟動thriftServer的時候，只要注意看日志就會發現。好了，今天就到這里，明天繼續玩~
參考文獻：《深入理解Spark核心思想與源碼解析》