《深入理解Spark：核心思想與原始碼分析》一書第一章的內容請看連結《第1章 環境準備》

《深入理解Spark：核心思想與原始碼分析》一書第二章的內容請看連結《第2章 SPARK設計理念與基本架構》

由於本書的第3章內容較多，所以打算分別開闢四篇隨筆分別展現。

本文展現第3章第三部分的內容：

3.8 TaskScheduler的啟動

　　3.7節介紹了任務排程器TaskScheduler的建立，要想TaskScheduler發揮作用，必須要啟動它，程式碼如下。

taskScheduler.start()

TaskScheduler在啟動的時候，實際呼叫了backend的start方法。

  override def start() {

    backend.start()

  }

以LocalBackend為例，啟動LocalBackend時向actorSystem註冊了LocalActor，見程式碼清單3-30所示（在《深入理解Spark：核心思想與原始碼分析》——SparkContext的初始化（中）》一文）。

3.8.1 建立LocalActor

　　建立LocalActor的過程主要是構建本地的Executor，見程式碼清單3-36。

程式碼清單3-36         LocalActor的實現

private[spark] class LocalActor(scheduler: TaskSchedulerImpl, executorBackend: LocalBackend,

  private val totalCores: Int) extends Actor with ActorLogReceive with Logging {

  import context.dispatcher   // to use Akka's scheduler.scheduleOnce()

  private var freeCores = totalCores

  private val localExecutorId = SparkContext.DRIVER_IDENTIFIER

  private val localExecutorHostname = "localhost"

 

  val executor = new Executor(

    localExecutorId, localExecutorHostname, scheduler.conf.getAll, totalCores, isLocal = true)

 

  override def receiveWithLogging = {

    case ReviveOffers =>

      reviveOffers()

 

    case StatusUpdate(taskId, state, serializedData) =>

      scheduler.statusUpdate(taskId, state, serializedData)

      if (TaskState.isFinished(state)) {

        freeCores += scheduler.CPUS_PER_TASK

        reviveOffers()

      }

 

    case KillTask(taskId, interruptThread) =>

      executor.killTask(taskId, interruptThread)

 

    case StopExecutor =>

      executor.stop()

  }

 

}

Executor的構建，見程式碼清單3-37，主要包括以下步驟：

1) 建立並註冊ExecutorSource。ExecutorSource是做什麼的呢？筆者將在3.10.2節詳細介紹。

2) 獲取SparkEnv。如果是非local模式，Worker上的CoarseGrainedExecutorBackend向Driver上的CoarseGrainedExecutorBackend註冊Executor時，則需要新建SparkEnv。可以修改屬性spark.executor.port（預設為0，表示隨機生成）來配置Executor中的ActorSystem的埠號。

3) 建立並註冊ExecutorActor。ExecutorActor負責接受傳送給Executor的訊息。

4) urlClassLoader的建立。為什麼需要建立這個ClassLoader？在非local模式中，Driver或者Worker上都會有多個Executor，每個Executor都設定自身的urlClassLoader，用於載入任務上傳的jar包中的類，有效對任務的類載入環境進行隔離。

5) 建立Executor執行TaskRunner任務（TaskRunner將在5.5節介紹）的執行緒池。此執行緒池是通過呼叫Utils.newDaemonCachedThreadPool建立的，具體實現請參閱附錄A。

6) 啟動Executor的心跳執行緒。此執行緒用於向Driver傳送心跳。

此外，還包括Akka傳送訊息的幀大小（10485760位元組）、結果總大小的位元組限制（1073741824位元組）、正在執行的task的列表、設定serializer的預設ClassLoader為建立的ClassLoader等。

程式碼清單3-37         Executor的構建

  val executorSource = new ExecutorSource(this, executorId)

  private val env = {

    if (!isLocal) {

      val port = conf.getInt("spark.executor.port", 0)

      val _env = SparkEnv.createExecutorEnv(

        conf, executorId, executorHostname, port, numCores, isLocal, actorSystem)

      SparkEnv.set(_env)

      _env.metricsSystem.registerSource(executorSource)

      _env.blockManager.initialize(conf.getAppId)

      _env

    } else {

      SparkEnv.get

    }

  }

 

  private val executorActor = env.actorSystem.actorOf(

    Props(new ExecutorActor(executorId)), "ExecutorActor")

 

  private val urlClassLoader = createClassLoader()

  private val replClassLoader = addReplClassLoaderIfNeeded(urlClassLoader)

  env.serializer.setDefaultClassLoader(urlClassLoader)

 

  private val akkaFrameSize = AkkaUtils.maxFrameSizeBytes(conf)

  private val maxResultSize = Utils.getMaxResultSize(conf)

 

  val threadPool = Utils.newDaemonCachedThreadPool("Executor task launch worker")

  private val runningTasks = new ConcurrentHashMap[Long, TaskRunner]

  startDriverHeartbeater()

3.8.2 ExecutorSource的建立與註冊

　　ExecutorSource用於測量系統。通過metricRegistry的register方法註冊計量，這些計量資訊包括threadpool.activeTasks、threadpool.completeTasks、threadpool.currentPool_size、threadpool.maxPool_size、filesystem.hdfs.write_bytes、filesystem.hdfs.read_ops、filesystem.file.write_bytes、filesystem.hdfs.largeRead_ops、filesystem.hdfs.write_ops等，ExecutorSource的實現見程式碼清單3-38。Metric介面的具體實現，參考附錄D。

程式碼清單3-38         ExecutorSource的實現

private[spark] class ExecutorSource(val executor: Executor, executorId: String) extends Source {

  private def fileStats(scheme: String) : Option[FileSystem.Statistics] =

    FileSystem.getAllStatistics().filter(s => s.getScheme.equals(scheme)).headOption

 

  private def registerFileSystemStat[T](

        scheme: String, name: String, f: FileSystem.Statistics => T, defaultValue: T) = {

    metricRegistry.register(MetricRegistry.name("filesystem", scheme, name), new Gauge[T] {

      override def getValue: T = fileStats(scheme).map(f).getOrElse(defaultValue)

    })

  }

  override val metricRegistry = new MetricRegistry()

  override val sourceName = "executor"

 

metricRegistry.register(MetricRegistry.name("threadpool", "activeTasks"), new Gauge[Int] {

    override def getValue: Int = executor.threadPool.getActiveCount()

  })

 metricRegistry.register(MetricRegistry.name("threadpool", "completeTasks"), new Gauge[Long] {

    override def getValue: Long = executor.threadPool.getCompletedTaskCount()

  })

  metricRegistry.register(MetricRegistry.name("threadpool", "currentPool_size"), new Gauge[Int] {

    override def getValue: Int = executor.threadPool.getPoolSize()

  })

  metricRegistry.register(MetricRegistry.name("threadpool", "maxPool_size"), new Gauge[Int] {

    override def getValue: Int = executor.threadPool.getMaximumPoolSize()

  })

 

  // Gauge for file system stats of this executor

  for (scheme <- Array("hdfs", "file")) {

    registerFileSystemStat(scheme, "read_bytes", _.getBytesRead(), 0L)

    registerFileSystemStat(scheme, "write_bytes", _.getBytesWritten(), 0L)

    registerFileSystemStat(scheme, "read_ops", _.getReadOps(), 0)

    registerFileSystemStat(scheme, "largeRead_ops", _.getLargeReadOps(), 0)

    registerFileSystemStat(scheme, "write_ops", _.getWriteOps(), 0)

  }

} 

建立完ExecutorSource後，呼叫MetricsSystem的registerSource方法將ExecutorSource註冊到MetricsSystem。registerSource方法使用MetricRegistry的register方法，將Source註冊到MetricRegistry，見程式碼清單3-39。關於MetricRegistry，具體參閱附錄D。

程式碼清單3-39         MetricsSystem註冊Source的實現

  def registerSource(source: Source) {

    sources += source

    try {

      val regName = buildRegistryName(source)

      registry.register(regName, source.metricRegistry)

    } catch {

      case e: IllegalArgumentException => logInfo("Metrics already registered", e)

    }

  } 

3.8.3 ExecutorActor的構建與註冊

　　ExecutorActor很簡單，當接收到SparkUI發來的訊息時，將所有執行緒的棧資訊傳送回去，程式碼實現如下。

  override def receiveWithLogging = {

    case TriggerThreadDump =>

      sender ! Utils.getThreadDump()

  }

3.8.4 Spark自身ClassLoader的建立

　　獲取要建立的ClassLoader的父載入器currentLoader，然後根據currentJars生成URL陣列，spark.files.userClassPathFirst屬性指定載入類時是否先從使用者的classpath下載入，最後建立ExecutorURLClassLoader或者ChildExecutorURLClassLoader，見程式碼清單3-40。

程式碼清單3-40         Spark自身ClassLoader的建立

  private def createClassLoader(): MutableURLClassLoader = {

    val currentLoader = Utils.getContextOrSparkClassLoader

 

    val urls = currentJars.keySet.map { uri =>

      new File(uri.split("/").last).toURI.toURL

    }.toArray

    val userClassPathFirst = conf.getBoolean("spark.files.userClassPathFirst", false)

    userClassPathFirst match {

      case true => new ChildExecutorURLClassLoader(urls, currentLoader)

      case false => new ExecutorURLClassLoader(urls, currentLoader)

    }

  }

Utils.getContextOrSparkClassLoader的實現見附錄A。ExecutorURLClassLoader或者ChildExecutorURLClassLoader實際上都繼承了URLClassLoader，見程式碼清單3-41。

程式碼清單3-41         ChildExecutorURLClassLoader與ExecutorURLClassLoader的實現

private[spark] class ChildExecutorURLClassLoader(urls: Array[URL], parent: ClassLoader)

  extends MutableURLClassLoader {

 

  private object userClassLoader extends URLClassLoader(urls, null){

    override def addURL(url: URL) {

      super.addURL(url)

    }

    override def findClass(name: String): Class[_] = {

      super.findClass(name)

    }

  }

 

  private val parentClassLoader = new ParentClassLoader(parent)

 

  override def findClass(name: String): Class[_] = {

    try {

      userClassLoader.findClass(name)

    } catch {

      case e: ClassNotFoundException => {

        parentClassLoader.loadClass(name)

      }

    }

  }

 

  def addURL(url: URL) {

    userClassLoader.addURL(url)

  }

 

  def getURLs() = {

    userClassLoader.getURLs()

  }

}

 

private[spark] class ExecutorURLClassLoader(urls: Array[URL], parent: ClassLoader)

  extends URLClassLoader(urls, parent) with MutableURLClassLoader {

 

  override def addURL(url: URL) {

    super.addURL(url)

  }

}

如果需要REPL互動，還會呼叫addReplClassLoaderIfNeeded建立replClassLoader，見程式碼清單3-42。

程式碼清單3-42         addReplClassLoaderIfNeeded的實現

  private def addReplClassLoaderIfNeeded(parent: ClassLoader): ClassLoader = {

    val classUri = conf.get("spark.repl.class.uri", null)

    if (classUri != null) {

      logInfo("Using REPL class URI: " + classUri)

      val userClassPathFirst: java.lang.Boolean =

        conf.getBoolean("spark.files.userClassPathFirst", false)

      try {

        val klass = Class.forName("org.apache.spark.repl.ExecutorClassLoader")

          .asInstanceOf[Class[_ <: ClassLoader]]

        val constructor = klass.getConstructor(classOf[SparkConf], classOf[String],

          classOf[ClassLoader], classOf[Boolean])

        constructor.newInstance(conf, classUri, parent, userClassPathFirst)

      } catch {

        case _: ClassNotFoundException =>

          logError("Could not find org.apache.spark.repl.ExecutorClassLoader on classpath!")

          System.exit(1)

          null

      }

    } else {

      parent

    }

  }

3.8.5 啟動Executor的心跳執行緒

　　Executor的心跳由startDriverHeartbeater啟動，見程式碼清單3-43。Executor心跳執行緒的間隔由屬性spark.executor.heartbeatInterval配置，預設是10000毫秒。此外，超時時間是30秒，超時重試次數是3次，重試間隔是3000毫秒，使用actorSystem.actorSelection (url)方法查詢到匹配的Actor引用， url是akka.tcp://[email protected] $driverHost:$driverPort/user/HeartbeatReceiver，最終建立一個執行過程中，每次會休眠10000到20000毫秒的執行緒。此執行緒從runningTasks獲取最新的有關Task的測量資訊，將其與executorId、blockManagerId封裝為Heartbeat訊息，向HeartbeatReceiver傳送Heartbeat訊息。

程式碼清單3-43         啟動Executor的心跳執行緒

  def startDriverHeartbeater() {

    val interval = conf.getInt("spark.executor.heartbeatInterval", 10000)

    val timeout = AkkaUtils.lookupTimeout(conf)

    val retryAttempts = AkkaUtils.numRetries(conf)

    val retryIntervalMs = AkkaUtils.retryWaitMs(conf)

    val heartbeatReceiverRef = AkkaUtils.makeDriverRef("HeartbeatReceiver", conf,env.actorSystem)

    val t = new Thread() {

      override def run() {

        // Sleep a random interval so the heartbeats don't end up in sync

        Thread.sleep(interval + (math.random * interval).asInstanceOf[Int])

        while (!isStopped) {

          val tasksMetrics = new ArrayBuffer[(Long, TaskMetrics)]()

          val curGCTime = gcTime

          for (taskRunner <- runningTasks.values()) {

            if (!taskRunner.attemptedTask.isEmpty) {

              Option(taskRunner.task).flatMap(_.metrics).foreach { metrics =>

                metrics.updateShuffleReadMetrics

                metrics.jvmGCTime = curGCTime - taskRunner.startGCTime

                if (isLocal) {

                  val copiedMetrics = Utils.deserialize[TaskMetrics](Utils.serialize(metrics))

                  tasksMetrics += ((taskRunner.taskId, copiedMetrics))

                } else {

                  // It will be copied by serialization

                  tasksMetrics += ((taskRunner.taskId, metrics))

                }

              }

            }

          }

          val message = Heartbeat(executorId, tasksMetrics.toArray, env.blockManager.blockManagerId)

          try {

            val response = AkkaUtils.askWithReply[HeartbeatResponse](message, heartbeatReceiverRef,

              retryAttempts, retryIntervalMs, timeout)

            if (response.reregisterBlockManager) {

              logWarning("Told to re-register on heartbeat")

              env.blockManager.reregister()

            }

          } catch {

            case NonFatal(t) => logWarning("Issue communicating with driver in heartbeater", t)

          }

          Thread.sleep(interval)

        }

      }

    }

    t.setDaemon(true)

    t.setName("Driver Heartbeater")

    t.start()

  }

這個心跳執行緒的作用是什麼呢？其作用有兩個：

q  更新正在處理的任務的測量資訊；

q  通知BlockManagerMaster，此Executor上的BlockManager依然活著。

下面對心跳執行緒的實現詳細分析下，讀者可以自行選擇是否需要閱讀。

　　初始化TaskSchedulerImpl後會建立心跳接收器HeartbeatReceiver。HeartbeatReceiver接受所有分配給當前Driver Application的Executor的心跳，並將Task、Task計量資訊、心跳等交給TaskSchedulerImpl和DAGScheduler作進一步處理。建立心跳接收器的程式碼如下。

  private val heartbeatReceiver = env.actorSystem.actorOf(

    Props(new HeartbeatReceiver(taskScheduler)), "HeartbeatReceiver")

HeartbeatReceiver在收到心跳訊息後，會呼叫TaskScheduler的executorHeartbeatReceived方法，程式碼如下。

  override def receiveWithLogging = {

    case Heartbeat(executorId, taskMetrics, blockManagerId) =>

      val response = HeartbeatResponse(

        !scheduler.executorHeartbeatReceived(executorId, taskMetrics, blockManagerId))

      sender ! response

  }

executorHeartbeatReceived的實現程式碼如下。

    val metricsWithStageIds: Array[(Long, Int, Int, TaskMetrics)] = synchronized {

      taskMetrics.flatMap { case (id, metrics) =>

        taskIdToTaskSetId.get(id)

          .flatMap(activeTaskSets.get)

          .map(taskSetMgr => (id, taskSetMgr.stageId, taskSetMgr.taskSet.attempt, metrics))

      }

    }

    dagScheduler.executorHeartbeatReceived(execId, metricsWithStageIds, blockManagerId)

這段程式通過遍歷taskMetrics，依據taskIdToTaskSetId和activeTaskSets找到TaskSetManager。然後將taskId、TaskSetManager.stageId、TaskSetManager .taskSet.attempt、TaskMetrics封裝到Array[(Long, Int, Int, TaskMetrics)]的陣列metricsWithStageIds中。最後呼叫了dagScheduler的executorHeartbeatReceived方法，其實現如下。

    listenerBus.post(SparkListenerExecutorMetricsUpdate(execId, taskMetrics))

    implicit val timeout = Timeout(600 seconds)

 

    Await.result(

      blockManagerMaster.driverActor ? BlockManagerHeartbeat(blockManagerId),

      timeout.duration).asInstanceOf[Boolean]

dagScheduler將executorId、metricsWithStageIds封裝為SparkListenerExecutorMetricsUpdate事件，並post到listenerBus中，此事件用於更新Stage的各種測量資料。最後給BlockManagerMaster持有的BlockManagerMasterActor傳送BlockManagerHeartbeat訊息。BlockManagerMasterActor在收到訊息後會匹配執行heartbeatReceived方法（會在4.3.1節介紹）。heartbeatReceived最終更新BlockManagerMaster對BlockManager最後可見時間（即更新BlockManagerId對應的BlockManagerInfo的_lastSeenMs，見程式碼清單3-44）。

程式碼清單3-44         BlockManagerMasterActor的心跳處理

private def heartbeatReceived(blockManagerId: BlockManagerId): Boolean = {

    if (!blockManagerInfo.contains(blockManagerId)) {

      blockManagerId.isDriver && !isLocal

    } else {

      blockManagerInfo(blockManagerId).updateLastSeenMs()

      true

    }

  }

local模式下Executor的心跳通訊過程，可以用圖3-3來表示。

圖3-3       Executor的心跳通訊過程

注意：在非local模式中Executor傳送心跳的過程是一樣的，主要的區別是Executor程序與Driver不在同一個程序，甚至不在同一個節點上。

接下來會初始化塊管理器BlockManager，程式碼如下。

env.blockManager.initialize(applicationId)

具體的初始化過程，請參閱第4章。

未完待續。。。

後記：自己犧牲了7個月的週末和下班空閒時間，通過研究Spark原始碼和原理，總結整理的《深入理解Spark：核心思想與原始碼分析》一書現在已經正式出版上市，目前亞馬遜、京東、噹噹、天貓等網站均有銷售，歡迎感興趣的同學購買。我開始研究原始碼時的Spark版本是1.2.0，經過7個多月的研究和出版社近4個月的流程，Spark自身的版本迭代也很快，如今最新已經是1.6.0。目前市面上另外2本原始碼研究的Spark書籍的版本分別是0.9.0版本和1.2.0版本，看來這些書的作者都與我一樣，遇到了這種問題。由於研究和出版都需要時間，所以不能及時跟上Spark的腳步，還請大家見諒。但是Spark核心部分的變化相對還是很少的，如果對版本不是過於追求，依然可以選擇本書。