《深入理解Spark：核心思想与源码分析》一书前言的内容请看链接《深入理解SPARK：核心思想与源码分析》一书正式出版上市

《深入理解Spark：核心思想与源码分析》一书第一章的内容请看链接《第1章 环境准备》

《深入理解Spark：核心思想与源码分析》一书第二章的内容请看链接《第2章 SPARK设计理念与基本架构》

由于本书的第3章内容较多，所以打算分别开辟三篇随笔分别展现。本文展现第3章第一部分的内容：

第3章 SparkContext的初始化

“道生一，一生二，二生三，三生万物。”——《道德经》

本章导读：

       SparkContext的初始化是Driver应用程序提交执行的前提，本章内容以local模式为主，并按照代码执行顺序讲解，这将有助于首次接触Spark的读者理解源码。读者朋友如果能边跟踪代码，边学习本章内容，也许是快速理解SparkContext初始化过程的便捷途径。已经熟练使用Spark的开发人员可以选择跳过本章内容。

       本章将在介绍SparkContext初始化过程的同时，向读者介绍各个组件的作用，为阅读后面的章节打好基础。Spark中的组件很多，就其功能而言涉及到网络通信、分布式、消息、存储、计算、缓存、测量、清理、文件服务、Web UI的方方面面。

3.1 SparkContext概述

        Spark Driver用于提交用户应用程序，实际可以看作Spark的客户端。了解Spark Driver的初始化，有助于读者理解用户应用程序在客户端的处理过程。

        Spark Driver的初始化始终围绕着SparkContext的初始化。SparkContext可以算得上是所有Spark应用程序的发动机引擎，轿车要想跑起来，发动机首先要启动。SparkContext初始化完毕，才能向Spark集群提交任务。在平坦的公路上，发动机只需以较低的转速，较低的功率就可以游刃有余；在山区，你可能需要一台能够提供大功率的发动机，这样才能满足你转山的体验。这些参数都是通过驾驶员操作油门、档位等传送给发动机的，而SparkContext的配置参数则由SparkConf负责，SparkConf就是你的操作面板。

SparkConf的构造很简单，主要是通过ConcurrentHashMap来维护各种Spark的配置属性。SparkConf代码结构见代码清单3-1。Spark的配置属性都是以“spark.”开头的字符串。

代码清单3-1  SparkConf代码结构

[java] view plain copy

1. class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging {  
2.   importSparkConf._  
3.   def this()= this(true)  
4.   private val settings = newConcurrentHashMap[String, String]()  
5.   if(loadDefaults) {  
6.     // 加载任何以spark.开头的系统属性  
7.     for ((key, value) <-Utils.getSystemProperties if key.startsWith("spark.")) {  
8.       set(key, value)  
9.     }  
10.   }  
11. //其余代码省略  

现在开始介绍SparkContext，SparkContext的初始化步骤如下：

1)        创建Spark执行环境SparkEnv；

2)        创建RDD清理器metadataCleaner；

3)        创建并初始化SparkUI；

4)        Hadoop相关配置及Executor环境变量的设置

5)        创建任务调度TaskScheduler；

6)        创建和启动DAGScheduler；

7)        TaskScheduler的启动；

8)        初始化块管理器BlockManager（BlockManager是存储体系的主要组件之一，将在第4章介绍）；

9)        启动测量系统MetricsSystem；

10)     创建和启动Executor分配管理器ExecutorAllocationManager；

11)     ContextCleaner的创建与启动；

12)     Spark环境更新；

13)     创建DAGSchedulerSource和BlockManagerSource；

14)     将SparkContext标记为激活。

SparkContext的主构造器参数为SparkConf，其实现如下。

[java] view plain copy

1. class SparkContext(config: SparkConf) extends Logging withExecutorAllocationClient {  
2. private val creationSite: CallSite = Utils.getCallSite()  
3.   private val allowMultipleContexts:Boolean =  
4.    config.getBoolean("spark.driver.allowMultipleContexts", false)  
5.  SparkContext.markPartiallyConstructed(this, allowMultipleContexts)  

上面代码中的CallSite存储了线程栈中最靠近栈顶的用户类及最靠近栈底的Scala或者Spark核心类信息。Utils.getCallSite的详细信息见附录A。SparkContext默认只有一个实例（由属性spark.driver.allowMultipleContexts来控制，用户需要多个SparkContext实例时，可以将其设置为true），方法markPartiallyConstructed用来确保实例的唯一性，并将当前SparkContext标记为正在构建中。

         接下来会对SparkConf进行拷贝，然后对各种配置信息进行校验，代码如下。

[java] view plain copy

1. private[spark] val conf =config.clone()  
2. conf.validateSettings()  
3.   
4. if (!conf.contains("spark.master")) {  
5.   throw newSparkException("A master URL must be set in your configuration")  
6. }  
7. if (!conf.contains("spark.app.name")) {  
8.   throw newSparkException("An application name must be set in yourconfiguration")  
9. }  

从上面校验的代码看到必须指定属性spark.master 和spark.app.name，否则会抛出异常，结束初始化过程。spark.master用于设置部署模式，spark.app.name指定应用程序名称。

3.2 创建执行环境SparkEnv

       SparkEnv是Spark的执行环境对象，其中包括众多与Executor执行相关的对象。由于在local模式下Driver会创建Executor，local-cluster部署模式或者Standalone部署模式下Worker另起的CoarseGrainedExecutorBackend进程中也会创建Executor，所以SparkEnv存在于Driver或者CoarseGrainedExecutorBackend进程中。创建SparkEnv 主要使用SparkEnv的createDriverEnv，createDriverEnv方法有三个参数，conf、isLocal和 listenerBus。

[java] view plain copy

1.  val isLocal = (master == "local" ||master.startsWith("local["))  
2.  private[spark] vallistenerBus = newLiveListenerBus  
3.  conf.set("spark.executor.id","driver")  
4.   
5.  private[spark] valenv =SparkEnv.createDriverEnv(conf,isLocal, listenerBus)  
6. SparkEnv.set(env)  

上面代码中的conf是对SparkConf的拷贝，isLocal标识是否是单机模式，listenerBus采用监听器模式维护各类事件的处理，在3.14节会详细介绍。

SparkEnv的方法createDriverEnv最终调用create创建SparkEnv。SparkEnv的构造步骤如下：

1)        创建安全管理器SecurityManager；

2)        创建基于Akka的分布式消息系统ActorSystem；

3)        创建Map任务输出跟踪器mapOutputTracker；

4)        实例化ShuffleManager；

5)        创建ShuffleMemoryManager；

6)        创建块传输服务BlockTransferService；

7)        创建BlockManagerMaster；

8)        创建块管理器BlockManager；

9)        创建广播管理器BroadcastManager；

10)    创建缓存管理器CacheManager；

11)    创建HTTP文件服务器HttpFileServer；

12)    创建测量系统MetricsSystem；

13)    创建SparkEnv；

3.2.1 安全管理器SecurityManager

         SecurityManager主要对权限、账号进行设置，如果使用Hadoop YARN作为集群管理器，则需要使用证书生成 secret key登录，最后给当前系统设置默认的口令认证实例，此实例采用匿名内部类实现，参见代码清单3-2。

代码清单3-2  SecurityManager的实现

[java] view plain copy

1. private val secretKey =generateSecretKey()  
2.   
3.  // 使用HTTP连接设置口令认证  
4.  if (authOn) {  
5.   Authenticator.setDefault(  
6.      newAuthenticator() {  
7.        override defgetPasswordAuthentication(): PasswordAuthentication = {  
8.          var passAuth:PasswordAuthentication = null  
9.          val userInfo =getRequestingURL().getUserInfo()  
10.          if (userInfo !=null) {  
11.            val  parts = userInfo.split(":",2)  
12.            passAuth = newPasswordAuthentication(parts(0),parts(1).toCharArray())  
13.          }  
14.          return passAuth  
15.        }  
16.      }  
17.    )  
18.  }<span style="font-family: Arial, Helvetica, sans-serif; background-color: rgb(255, 255, 255);"> </span>  

3.2.2 基于Akka的分布式消息系统ActorSystem

         ActorSystem是Spark中最基础的设施，Spark既使用它发送分布式消息，又用它实现并发编程。怎么，消息系统可以实现并发？要解释清楚这个问题，首先应该简单的介绍下Scala语言的Actor并发编程模型：Scala认为Java线程通过共享数据以及通过锁来维护共享数据的一致性是糟糕的做法，容易引起锁的争用，而且线程的上下文切换会带来不少开销，降低并发程序的性能，甚至会引入死锁的问题。在Scala中只需要自定义类型继承Actor，并且提供act方法，就如同Java里实现Runnable接口，需要实现run方法一样。但是不能直接调用act方法，而是通过发送消息的方式(Scala发送消息是异步的)，传递数据。如：

         Actor ! message

         Akka是Actor编程模型的高级类库，类似于JDK 1.5之后越来越丰富的并发工具包，简化了程序员并发编程的难度。ActorSystem便是Akka提供的用于创建分布式消息通信系统的基础类。Akka的具体信息见附录B。

         正式因为Actor轻量级的并发编程、消息发送以及ActorSystem支持分布式消息发送等特点，Spark选择了ActorSystem。

         SparkEnv中创建ActorSystem时用到了AkkaUtils工具类，见代码清单3-3。AkkaUtils.createActorSystem方法用于启动ActorSystem，见代码清单3-4。AkkaUtils使用了Utils的静态方法startServiceOnPort， startServiceOnPort最终会回调方法startService: Int => (T, Int)，此处的startService实际是方法doCreateActorSystem。真正启动ActorSystem是由doCreateActorSystem方法完成的，doCreateActorSystem的具体实现细节请见附录B。Spark的Driver中Akka的默认访问地址是akka://sparkDriver，Spark的Executor中Akka的默认访问地址是akka://sparkExecutor。如果不指定ActorSystem的端口，那么所有节点的ActorSystem端口在每次启动时随机产生。关于startServiceOnPort的实现，请见附录A。

代码清单3-3  使用AkkaUtils工具类创建和启动[计算机3] [初霖4] ActorSystem

[java] view plain copy

1. val(actorSystem, boundPort) =  
2.  Option(defaultActorSystem) match {  
3.     case Some(as)=> (as, port)  
4.     case None =>  
5.       valactorSystemName =if (isDriver) driverActorSystemNameelse executorActorSystemName  
6.      AkkaUtils.createActorSystem(actorSystemName, hostname, port, conf,securityManager)  
7.   }  

代码清单3-4  ActorSystem的创建和启动

[java] view plain copy

1. def createActorSystem(  
2.     name:String,  
3.     host:String,  
4.     port:Int,  
5.     conf:SparkConf,  
6.     securityManager: SecurityManager):(ActorSystem, Int) = {  
7.   val startService: Int=> (ActorSystem, Int) = { actualPort =>  
8.    doCreateActorSystem(name, host, actualPort, conf, securityManager)  
9.   }  
10.  Utils.startServiceOnPort(port, startService, conf, name)  
11. }  

3.2.3 map任务输出跟踪器mapOutputTracker

         mapOutputTracker用于跟踪map阶段任务的输出状态，此状态便于reduce阶段任务获取地址及中间输出结果。每个map任务或者reduce任务都会有其唯一标识，分别为mapId和reduceId。每个reduce任务的输入可能是多个map任务的输出，reduce会到各个map任务的所在节点上拉取Block，这一过程叫做shuffle。每批shuffle过程都有唯一的标识shuffleId。

         这里先介绍下MapOutputTrackerMaster。MapOutputTrackerMaster内部使用mapStatuses：TimeStampedHashMap[Int,Array[MapStatus]]来维护跟踪各个map任务的输出状态。其中key对应shuffleId，Array存储各个map任务对应的状态信息MapStatus。由于MapStatus维护了map输出Block的地址BlockManagerId，所以reduce任务知道从何处获取map任务的中间输出。MapOutputTrackerMaster还使用cachedSerializedStatuses：TimeStampedHashMap[Int, Array[Byte]]维护序列化后的各个map任务的输出状态。其中key对应shuffleId，Array存储各个序列化MapStatus生成的字节数组。

         Driver和Executor处理MapOutputTrackerMaster的方式有所不同：

• 如果当前应用程序是Driver，则创建MapOutputTrackerMaster，然后创建MapOutputTrackerMasterActor，并且注册到ActorSystem中。
• 如果当前应用程序是Executor，则创建MapOutputTrackerWorker，并从ActorSystem中找到MapOutputTrackerMasterActor。

无论是Driver还是Executor，最后都由mapOutputTracker的属性trackerActor持有MapOutputTrackerMasterActor的引用，参见代码清单3-5。

代码清单3-5  registerOrLookup方法用于查找或者注册Actor的实现

[java] view plain copy

1. def registerOrLookup(name: String, newActor: => Actor): ActorRef ={  
2.       if (isDriver) {  
3.        logInfo("Registering" + name)  
4.         actorSystem.actorOf(Props(newActor),name = name)  
5.       } else {  
6.        AkkaUtils.makeDriverRef(name, conf, actorSystem)  
7.       }  
8.     }  
9.    
10.     val mapOutputTracker=  if (isDriver) {  
11.       newMapOutputTrackerMaster(conf)  
12.     } else {  
13.       newMapOutputTrackerWorker(conf)  
14. }  
15.    
16.     mapOutputTracker.trackerActor= registerOrLookup(  
17.      "MapOutputTracker",  
18.       newMapOutputTrackerMasterActor(mapOutputTracker.asInstanceOf[MapOutputTrackerMaster], conf))  

在后面章节大家会知道map任务的状态正是由Executor向持有的MapOutputTrackerMasterActor发送消息，将map任务状态同步到mapOutputTracker的mapStatuses和cachedSerializedStatuses的。Executor究竟是如何找到MapOutputTrackerMasterActor的？registerOrLookup方法通过调用AkkaUtils.makeDriverRef找到MapOutputTrackerMasterActor，实际正是利用ActorSystem提供的分布式消息机制实现的，具体细节参见附录B。这里第一次使用到了Akka提供的功能，以后大家会渐渐感觉到使用Akka的便捷。

3.2.4 实例化ShuffleManager

         ShuffleManager负责管理本地及远程的block数据的shuffle操作。ShuffleManager默认为通过反射方式生成的SortShuffleManager的实例，可以修改属性spark.shuffle.manager为hash来显式[计算机5] [初霖6] 使用HashShuffleManager。SortShuffleManager通过持有的IndexShuffleBlockManager间接操作BlockManager中的DiskBlockManager将map结果写入本地，并根据shuffleId、mapId写入索引文件，也能通过MapOutputTrackerMaster中维护的mapStatuses从本地或者其他远程节点读取文件。有读者可能会问，为什么需要shuffle？Spark作为并行计算框架，同一个作业会被划分为多个任务在多个节点上并行执行，reduce的输入可能存在于多个节点上，因此需要通过“洗牌”将所有reduce的输入汇总起来，这个过程就是shuffle。这个问题以及对ShuffleManager的具体使用会在第5章和第6章详述。ShuffleManager的实例化见代码清单3-6。代码清单3-6最后创建的ShuffleMemoryManager，将在3.2.5节介绍。

代码清单3-6  ShuffleManager的实例化及ShuffleMemoryManager的创建

[java] view plain copy

1. val shortShuffleMgrNames =Map(  
2.   "hash"-> "org.apache.spark.shuffle.hash.HashShuffleManager",  
3.   "sort"-> "org.apache.spark.shuffle.sort.SortShuffleManager")  
4. val shuffleMgrName = conf.get("spark.shuffle.manager", "sort")  
5. val shuffleMgrClass = shortShuffleMgrNames.get  
6. se(shuffleMgrName.toLowerCase, shuffleMgrName)  
7. val shuffleManager = instantiateClass[ShuffleManager](shuffleMgrClass)  
8.   
9. val shuffleMemoryManager =new ShuffleMemoryManager(conf)  

3.2.5 shuffle线程内存管理器ShuffleMemoryManager

         ShuffleMemoryManager负责管理shuffle线程占有内存的分配与释放，并通过threadMemory：mutable.HashMap[Long, Long]缓存每个线程的内存字节数，见代码清单3-7。

代码清单3-7  ShuffleMemoryManager的数据结构

[java] view plain copy

1. private[spark] class ShuffleMemoryManager(maxMemory: Long)extends Logging {  
2.   private val threadMemory = newmutable.HashMap[Long, Long]() // threadId -> memory bytes  
3.   def this(conf: SparkConf) = this(ShuffleMemoryManager.getMaxMemory(conf))  

getMaxMemory方法用于获取shuffle所有线程占用的最大内存，实现如下。

[java] view plain copy

1. def getMaxMemory(conf: SparkConf): Long = {  
2.     val memoryFraction =conf.getDouble("spark.shuffle.memoryFraction", 0.2)  
3.     val safetyFraction =conf.getDouble("spark.shuffle.safetyFraction", 0.8)  
4.    (Runtime.getRuntime.maxMemory * memoryFraction *safetyFraction).toLong  
5.   }  

从上面代码可以看出，shuffle所有线程占用的最大内存的计算公式为：

Java运行时最大内存 * Spark的shuffle最大内存占比 * Spark的安全内存占比

可以配置属性spark.shuffle.memoryFraction修改Spark的shuffle最大内存占比，配置属性spark.shuffle.safetyFraction修改Spark的安全内存占比。

注意：ShuffleMemoryManager通常运行在Executor中， Driver中的ShuffleMemoryManager 只有在local模式下才起作用。

3.2.6 块传输服务BlockTransferService

         BlockTransferService默认为NettyBlockTransferService（可以配置属性spark.shuffle.blockTransferService使用NioBlockTransferService），它使用Netty提供的异步事件驱动的网络应用框架，提供web服务及客户端，获取远程节点上Block的集合。

[java] view plain copy

1. val blockTransferService =  
2.      conf.get("spark.shuffle.blockTransferService","netty").toLowerCase match {  
3.         case "netty"=>  
4.           newNettyBlockTransferService(conf, securityManager, numUsableCores)  
5.         case "nio"=>  
6.           newNioBlockTransferService(conf, securityManager)  
7.       }  

NettyBlockTransferService的具体实现将在第4章详细介绍。这里大家可能觉得奇怪，这样的网络应用为何也要放在存储体系？大家不妨先带着疑问，直到你真正了解存储体系。

3.2.7 BlockManagerMaster介绍

BlockManagerMaster负责对Block的管理和协调，具体操作依赖于BlockManagerMasterActor。Driver和Executor处理BlockManagerMaster的方式不同：

• 如果当前应用程序是Driver，则创建BlockManagerMasterActor，并且注册到ActorSystem中。
• 如果当前应用程序是Executor，则从ActorSystem中找到BlockManagerMasterActor。

无论是Driver还是Executor，最后BlockManagerMaster的属性driverActor将持有对BlockManagerMasterActor的引用。BlockManagerMaster的创建代码如下。

[java] view plain copy

1. val blockManagerMaster = new BlockManagerMaster(registerOrLookup(  
2.      "BlockManagerMaster",  
3.       newBlockManagerMasterActor(isLocal, conf, listenerBus)), conf, isDriver)  

registerOrLookup已在3.2.3节介绍过了，不再赘述。BlockManagerMaster及BlockManagerMasterActor的具体实现将在第4章详细介绍。

3.2.8 创建块管理器BlockManager

         BlockManager负责对Block的管理，只有在BlockManager的初始化方法initialize被调用后，它才是有效的。BlockManager作为存储系统的一部分，具体实现见第4章。BlockManager的创建代码如下。

[java] view plain copy

1. val blockManager = new BlockManager(executorId, actorSystem, blockManagerMaster,  
2.      serializer, conf, mapOutputTracker, shuffleManager,blockTransferService, securityManager,  
3.       numUsableCores)  

3.2.9 创建广播管理器BroadcastManager

         BroadcastManager用于将配置信息和序列化后的RDD、Job以及ShuffleDependency等信息在本地存储。如果为了容灾，也会复制到其他节点上。创建BroadcastManager的代码实现如下。

[java] view plain copy

1. val broadcastManager = new BroadcastManager(isDriver, conf,securityManager)  

BroadcastManager必须在其初始化方法initialize被调用后，才能生效。Initialize方法实际利用反射生成广播工厂实例broadcastFactory（可以配置属性spark.broadcast.factory指定，默认为org.apache.spark.broadcast.TorrentBroadcastFactory）。BroadcastManager的广播方法newBroadcast实际代理了工厂broadcastFactory的newBroadcast方法来生成广播或者非广播对象。BroadcastManager的Initialize及newBroadcast方法见代码清单3-8。

代码清单3-8  BroadcastManager的实现

[java] view plain copy

1.   private def initialize(){  
2.    synchronized {  
3.       if(!initialized) {  
4.         val broadcastFactoryClass = conf.get("spark.broadcast.factory","org.apache.spark.broadcast.TorrentBroadcastFactory")  
5.        broadcastFactory =  
6.          Class.forName(broadcastFactoryClass).newInstance.asInstanceOf[BroadcastFactory]  
7.        broadcastFactory.initialize(isDriver, conf, securityManager)  
8.        initialized = true  
9.       }  
10.     }  
11.   }  
12.    
13.   private val nextBroadcastId = new AtomicLong(0)  
14.    
15.   defnewBroadcast[T: ClassTag](value_ : T, isLocal: Boolean) = {  
16.    broadcastFactory.newBroadcast[T](value_, isLocal, nextBroadcastId.getAndIncrement())  
17.   }  
18.    
19.   defunbroadcast(id: Long, removeFromDriver: Boolean, blocking: Boolean) {  
20.    broadcastFactory.unbroadcast(id, removeFromDriver, blocking)  
21.   }  
22. }<span style="font-family: Arial, Helvetica, sans-serif; background-color: rgb(255, 255, 255);"> </span>  

3.2.10 创建缓存管理器CacheManager

         CacheManager用于缓存RDD某个分区计算后中间结果，缓存计算结果发生在迭代计算的时候，将在6.1节讲到。而CacheManager将在4.14节详细描述。创建CacheManager的代码如下。

[java] view plain copy

1. val cacheManager = new CacheManager(blockManager)  

3.2.11 HTTP文件服务器HttpFileServer

         参见代码清单3-9。HttpFileServer主要提供对jar及其他文件的http访问，这些jar包包括用户上传的jar包。端口由属性spark.fileserver.port配置，默认为0，表示随机生成端口号。

代码清单3-9  HttpFileServer的创建

[java] view plain copy

1. val httpFileServer =  
2.     if (isDriver) {  
3.       val fileServerPort = conf.getInt("spark.fileserver.port",0)  
4.       val server = newHttpFileServer(conf, securityManager, fileServerPort)  
5.      server.initialize()  
6.      conf.set("spark.fileserver.uri",  server.serverUri)  
7.      server  
8.     } else {  
9.       null  
10.     }  

HttpFileServer的初始化过程，见代码清单3-10，主要包括以下步骤：

1)        使用Utils工具类创建文件服务器的根目录及临时目录（临时目录在运行时环境关闭时会删除）。Utils工具的详细介绍，见附录A。

2)        创建存放jar包及其他文件的文件目录。

3)        创建并启动HTTP服务。

代码清单3-10         HttpFileServer的初始化

[java] view plain copy

1. def initialize(){  
2.   baseDir= Utils.createTempDir(Utils.getLocalDir(conf), "httpd")  
3.   fileDir= new File(baseDir,"files")  
4.   jarDir =new File(baseDir,"jars")  
5.  fileDir.mkdir()  
6.  jarDir.mkdir()  
7.  logInfo("HTTP File server directory is " + baseDir)  
8.  httpServer = new HttpServer(conf, baseDir, securityManager, requestedPort,"HTTP file server")  
9.  httpServer.start()  
10.  serverUri = httpServer.uri  
11.  logDebug("HTTP file server started at: " + serverUri)  
12. }  

HttpServer的构造和start方法的实现中，再次使用了Utils的静态方法startServiceOnPort，因此会回调doStart方法，见代码清单3-11。有关jetty的API使用参见附录C。

代码清单3-11         HttpServer的启动

[java] view plain copy

1. def start() {  
2.   if (server != null) {  
3.     throw newServerStateException("Serveris already started")  
4.   } else {  
5.    logInfo("Starting HTTP Server")  
6.     val(actualServer, actualPort) =  
7.      Utils.startServiceOnPort[Server](requestedPort, doStart, conf,serverName)  
8.     server= actualServer  
9.     port =actualPort  
10.   }  
11. }  

doStart方法中启动内嵌的jetty所提供的HTTP服务，见代码清单3-12。

代码清单3-12         HttpServer的启动功能实现

[java] view plain copy

1. private def doStart(startPort: Int): (Server, Int) = {  
2.    val server = new Server()  
3.    val connector = newSocketConnector  
4.    connector.setMaxIdleTime(60 *1000)  
5.    connector.setSoLingerTime(-1)  
6.    connector.setPort(startPort)  
7.    server.addConnector(connector)  
8.   
9.    val threadPool = newQueuedThreadPool  
10.    threadPool.setDaemon(true)  
11.    server.setThreadPool(threadPool)  
12.    val resHandler = newResourceHandler  
13.    resHandler.setResourceBase(resourceBase.getAbsolutePath)  
14.   
15.    val handlerList = new HandlerList  
16.    handlerList.setHandlers(Array(resHandler,newDefaultHandler))  
17.   
18.    if(securityManager.isAuthenticationEnabled()) {  
19.     logDebug("HttpServer is using security")  
20.      val sh =setupSecurityHandler(securityManager)  
21.      //make sure we go through security handler to get resources  
22.      sh.setHandler(handlerList)  
23.      server.setHandler(sh)  
24.    } else {  
25.     logDebug("HttpServer is not using security")  
26.      server.setHandler(handlerList)  
27.    }  
28.   
29.    server.start()  
30.    val actualPort = server.getConnectors()(0).getLocalPort  
31.   
32.    (server,actualPort)  
33.  }  

3.2.12 创建测量系统MetricsSystem

         MetricsSystem是Spark的测量系统，创建MetricsSystem的代码如下。

[java] view plain copy

1. val metricsSystem = if (isDriver) {  
2.    MetricsSystem.createMetricsSystem("driver", conf,securityManager)  
3.   } else {  
4.    conf.set("spark.executor.id", executorId)  
5.     val ms = MetricsSystem.createMetricsSystem("executor",conf,securityManager)  
6.     ms.start()  
7.     ms  
8.   }  

上面调用的createMetricsSystem方法实际创建了MetricsSystem，代码如下。

[java] view plain copy

1. def createMetricsSystem(  
2.    instance: String, conf: SparkConf, securityMgr: SecurityManager): MetricsSystem= {  
3.   new MetricsSystem(instance, conf, securityMgr)  
4. }  

构造MetricsSystem的过程最重要的是调用了MetricsConfig的initialize方法，见代码清单3-13。

代码清单3-13         MetricsConfig的初始化

[java] view plain copy

1. def initialize() {  
2.    setDefaultProperties(properties)  
3.    
4.     var is: InputStream = null  
5.     try {  
6.       is = configFilematch {  
7.         case Some(f) =>newFileInputStream(f)  
8.         case None =>Utils.getSparkClassLoader.getResourceAsStream(METRICS_CONF)  
9.       }  
10.    
11.       if (is !=null) {  
12.         properties.load(is)  
13.       }  
14.     } catch {  
15.       case e: Exception =>logError("Error loadingconfigure file",e)  
16.     } finally {  
17.       if (is !=null) is.close()  
18.     }  
19.    
20.    propertyCategories = subProperties(properties,INSTANCE_REGEX)  
21.     if(propertyCategories.contains(DEFAULT_PREFIX)) {  
22.       import scala.collection.JavaConversions._  
23.    
24.       val defaultProperty =propertyCategories(DEFAULT_PREFIX)  
25.       for { (inst,prop) <-propertyCategories  
26.             if (inst !=DEFAULT_PREFIX)  
27.            (k, v) <- defaultProperty  
28.             if (prop.getProperty(k) ==null) } {  
29.         prop.setProperty(k,v)  
30.       }  
31.     }  
32.   }  

从以上实现可以看出，MetricsConfig的initialize方法主要负责加载metrics.properties文件中的属性配置，并对属性进行初始化转换。

例如：将属性

{*.sink.servlet.path=/metrics/json,applications.sink.servlet.path=/metrics/applications/json,*.sink.servlet.class=org.apache.spark.metrics.sink.MetricsServlet, master.sink.servlet.path=/metrics/master/json}

转换为

Map(applications ->{sink.servlet.class=org.apache.spark.metrics.sink.MetricsServlet,sink.servlet.path=/metrics/applications/json}, master ->{sink.servlet.class=org.apache.spark.metrics.sink.MetricsServlet, sink.servlet.path=/metrics/master/json},* -> {sink.servlet.class=org.apache.spark.metrics.sink.MetricsServlet,sink.servlet.path=/metrics/json})

3.2.13 创建SparkEnv

         当所有的基础组件准备好后，最终使用下面的代码创建执行环境SparkEnv。

[java] view plain copy

1. new SparkEnv(executorId, actorSystem, serializer, closureSerializer, cacheManager,  
2.      mapOutputTracker, shuffleManager, broadcastManager, blockTransferService,  
3.  blockManager,securityManager, httpFileServer, sparkFilesDir,  
4. metricsSystem, shuffleMemoryManager, conf)  

注意：serializer和closureSerializer都是使用Class.forName反射生成的org.apache.spark.serializer.JavaSerializer类的实例，其中closureSerializer实例特别用来对Scala中的闭包进行序列化。

3.3 创建metadataCleaner

SparkContext为了保持对所有持久化的RDD的跟踪，使用类型是TimeStampedWeakValueHashMap的persistentRdds缓存。metadataCleaner的功能是清除过期的持久化RDD。创建metadataCleaner的代码如下。

[java] view plain copy

1. private[spark] val persistentRdds = newTimeStampedWeakValueHashMap[Int, RDD[_]]  
2. private[spark] val metadataCleaner =  
3.   new MetadataCleaner(MetadataCleanerType.SPARK_CONTEXT,this.cleanup, conf)  

我们仔细看看MetadataCleaner的实现，见代码清单3-14。

代码清单3-14         MetadataCleaner的实现

[java] view plain copy

1. private[spark] class MetadataCleaner(  
2.    cleanerType: MetadataCleanerType.MetadataCleanerType,  
3.    cleanupFunc: (Long) => Unit,  
4.     conf:SparkConf)  
5.   extends Logging  
6. {  
7.   val name =cleanerType.toString  
8.    
9.   private val delaySeconds =MetadataCleaner.getDelaySeconds(conf, cleanerType)  
10.   private valperiodSeconds = math.max(10, delaySeconds /10)  
11.   private val timer = new Timer(name +" cleanup timer", true)  
12.    
13.   private val task = new TimerTask {  
14.     override def run() {  
15.       try {  
16.         cleanupFunc(System.currentTimeMillis()- (delaySeconds *1000))  
17.        logInfo("Ran metadata cleaner for " + name)  
18.       } catch {  
19.         case e: Exception =>logError("Error runningcleanup task for " +name, e)  
20.       }  
21.     }  
22.   }  
23.    
24.   if (delaySeconds> 0){  
25.    timer.schedule(task, delaySeconds * 1000, periodSeconds *1000)  
26.   }  
27.    
28.   def cancel() {  
29.     timer.cancel()  
30.   }  
31. }  

从MetadataCleaner的实现可以看出其实质是一个用TimerTask实现的定时器，不断调用cleanupFunc: (Long) => Unit这样的函数参数。构造metadataCleaner时的函数参数是cleanup，用于清理persistentRdds中的过期内容，代码如下。

[java] view plain copy

1. private[spark] defcleanup(cleanupTime: Long) {  
2.   persistentRdds.clearOldValues(cleanupTime)  
3. }  

 未完待续。。。

后记：自己牺牲了7个月的周末和下班空闲时间，通过研究Spark源码和原理，总结整理的《深入理解Spark：核心思想与源码分析》一书现在已经正式出版上市，目前亚马逊、京东、当当、天猫等网站均有销售，欢迎感兴趣的同学购买。我开始研究源码时的Spark版本是1.2.0，经过7个多月的研究和出版社近4个月的流程，Spark自身的版本迭代也很快，如今最新已经是1.6.0。目前市面上另外2本源码研究的Spark书籍的版本分别是0.9.0版本和1.2.0版本，看来这些书的作者都与我一样，遇到了这种问题。由于研究和出版都需要时间，所以不能及时跟上Spark的脚步，还请大家见谅。但是Spark核心部分的变化相对还是很少的，如果对版本不是过于追求，依然可以选择本书。

京东（现有满100减30活动）：http://item.jd.com/11846120.html 

当当：http://product.dangdang.com/23838168.html