Spark源码阅读3——SparkContext初始化流程

2019-05-27

SparkContext是Spark函数的主入口，一个SparkContext代表与Spark集群的连接，可以用来在集群创建RDD、累加器、广播变量等。每一个JVM只能有一个活跃的SparkContext实例，在创建新的SparkContext实例之前，必须stop正活跃的SparkContext实例。不过这个限制最终可能会移除掉。

1. SparkContext类初始化

纵观SpartContext类的初始化过程，主要创建了LiveListenerBus、SparkStatusTracker、HeartbeatReceiver、SchedulerBackend、TaskScheduler和DAGScheduler实例。这些实例，在后期会完成event监听、status追踪、heartbeat接受、后台调度、任务调度、DAG调度等工作，是Spark Application得以运行的基石。

private var _conf: SparkConf = _
private var _eventLogDir: Option[URI] = None
private var _eventLogCodec: Option[String] = None
private var _env: SparkEnv = _
private var _jobProgressListener: JobProgressListener = _
private var _statusTracker: SparkStatusTracker = _
private var _progressBar: Option[ConsoleProgressBar] = None
private var _ui: Option[SparkUI] = None
private var _hadoopConfiguration: Configuration = _
private var _executorMemory: Int = _
private var _schedulerBackend: SchedulerBackend = _
private var _taskScheduler: TaskScheduler = _
private var _heartbeatReceiver: RpcEndpointRef = _
@volatile private var _dagScheduler: DAGScheduler = _
private var _applicationId: String = _
private var _applicationAttemptId: Option[String] = None
private var _eventLogger: Option[EventLoggingListener] = None
private var _executorAllocationManager: Option[ExecutorAllocationManager] = None
private var _cleaner: Option[ContextCleaner] = None
private var _listenerBusStarted: Boolean = false
private var _jars: Seq[String] = _
private var _files: Seq[String] = _
private var _shutdownHookRef: AnyRef = _

一些变量的初始化，包括新建LiveListenerBus类实例listerBus。
复制SparkConf config到_conf，并对其中的设置进行验证，检查是否存在非法或弃用的参数。
若_conf中不包含spark.master或spark.app.name则抛出异常。
在yarn cluster模式下，_conf必须包含spark.yarn.app.id，否则抛出异常。
在_conf中设置spark.driver.host和spark.executor.id=driver。
新建 JobProgressListener类实例_jobProgressListener，并执行listenerBus.addListener(jobProgressListener)。
调用createSparkEnv(_conf,isLocal,listenerBus)方法，创建Driver的SparkEnv实例。SparkEnv类包含了运行Spark实例(master/worker)的运行时环境，包含serializer、block manager等。

class SparkEnv (
    val executorId: String,
    private[spark] val rpcEnv: RpcEnv,
    val serializer: Serializer,
    val closureSerializer: Serializer,
    val serializerManager: SerializerManager,
    val mapOutputTracker: MapOutputTracker,
    val shuffleManager: ShuffleManager,
    val broadcastManager: BroadcastManager,
    val blockManager: BlockManager,
    val securityManager: SecurityManager,
    val metricsSystem: MetricsSystem,
    val memoryManager: MemoryManager,
    val outputCommitCoordinator: OutputCommitCoordinator,
    val conf: SparkConf) extends Logging {}

创建SparkStatusTracker实例。
初始化hadoop配置变量_hadoopConfiguration = SparkHadoopUtil.get.newConfiguration(_conf)。
调用jars.foreach(addJar)，为所有的任务添加 jar dependency。
调用files.foreach(addFIle)，为每个node 加载文件。
设置_executorMemory，为executor设置内存。
_heartbeatReceiver=env.rpcEnv.setupEndpoint(HeartbeatReceiver.ENDPOINT_NAME, new HeartbeatReceiver(this))，创建HeartbeatReceiver类实例。
创建SchedulerBackend、TaskScheduler、DAGScheduler类实例。

val (sched, ts) = SparkContext.createTaskScheduler(this, master, deployMode)
  _schedulerBackend = sched
  _taskScheduler = ts
  _dagScheduler = new DAGScheduler(this)
  _heartbeatReceiver.ask[Boolean](TaskSchedulerIsSet)

启动_taskScheduler.start()。
设置spark.app.id。
启动_env.metricsSystem.start()，并将driver metrics servlet handler 附加到 web ui。
创建ContextCleaner类实例，启动cleaner。
调用postEnvironmentUpdate()方法，通过listenerBus发送envirenmentUpdate event。
调用postApplicationStart()方法，通过listenerBus发送application start event。该方法假设TaskScheduler已经初始化并已经让cluster manager获取到了application ID。

/** Post the application start event */
private def fpostApplicationStart() {
  // Note: this code assumes that the task scheduler has been initialized and has contacted
  // the cluster manager to get an application ID (in case the cluster manager provides one).
  listenerBus.post(SparkListenerApplicationStart(appName, Some(applicationId),
    startTime, sparkUser, applicationAttemptId, schedulerBackend.getDriverLogUrls))
}

调用_taskScheduler.postStartHook()方法，当系统初始化成功后，会调用该方法，Yarn使用该方法引导基于机架感知、等待从机注册的资源分配。

2. createTaskScheduler(sc, master, deployMode)方法

那么，SparkContext是在哪一步分配Executor呢，在步骤14中，val (sched, ts)=SparkContext.createTaskScheduler(this, master, deployMode)命令，会根据不同的master, deployMode，启动不同的SchedulerBackend和TaskScheduler。

/**
 * Create a task scheduler based on a given master URL.
 * Return a 2-tuple of the scheduler backend and the task scheduler.
 */
private def createTaskScheduler(
    sc: SparkContext,
    master: String,
    deployMode: String): (SchedulerBackend, TaskScheduler) = {
  import SparkMasterRegex._

  // When running locally, don't try to re-execute tasks on failure.
  val MAX_LOCAL_TASK_FAILURES = 1

  master match {
    case "local" =>
      val scheduler = new TaskSchedulerImpl(sc, MAX_LOCAL_TASK_FAILURES, isLocal = true)
      val backend = new LocalSchedulerBackend(sc.getConf, scheduler, 1)
      scheduler.initialize(backend)
      (backend, scheduler)

    case LOCAL_N_REGEX(threads) =>
      def localCpuCount: Int = Runtime.getRuntime.availableProcessors()
      // local[*] estimates the number of cores on the machine; local[N] uses exactly N threads.
      val threadCount = if (threads == "*") localCpuCount else threads.toInt
      if (threadCount <= 0) {
        throw new SparkException(s"Asked to run locally with $threadCount threads")
      }
      val scheduler = new TaskSchedulerImpl(sc, MAX_LOCAL_TASK_FAILURES, isLocal = true)
      val backend = new LocalSchedulerBackend(sc.getConf, scheduler, threadCount)
      scheduler.initialize(backend)
      (backend, scheduler)

  ......
  ......
 
    case masterUrl =>
      val cm = getClusterManager(masterUrl) match {
        case Some(clusterMgr) => clusterMgr
        case None => throw new SparkException("Could not parse Master URL: '" + master + "'")
      }
      try {
        val scheduler = cm.createTaskScheduler(sc, masterUrl)
        val backend = cm.createSchedulerBackend(sc, masterUrl, scheduler)
        cm.initialize(scheduler, backend)
        (backend, scheduler)
      } catch {
        case se: SparkException => throw se
        case NonFatal(e) =>
          throw new SparkException("External scheduler cannot be instantiated", e)
      }
  }
}

从代码中我们可以发现，该方法会根据master和depolyMode的不同，来构造不同的SchedulerBackend和TaskScheduler。从而实现了，Spark可以适配local、standalone、yarn等多种不同模式。Yarn Cluster模式下会实例化CoarseGrainedSchedulerBackend类。该类会持有Executor资源。

3. CoarseGrainedSchedulerBackend类

CoarseGrainedSchedulerBackend类的start()方法会被TaskSchedulerImpl类调用，在start()方法中，会实例化DriverEndpoint。

在DriverEndpoint类的receiverAndReply方法中，driver会根据收到RegisterExecutor、StopDriver、StopExecutors、RemoveExecutor、RetrieveSparkAppConfig等不同的命令，进行相应的操作。如会根据接收到的RegisterExecutor RPC命令，完成Executor的注册。

override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {

      case RegisterExecutor(executorId, executorRef, hostname, cores, logUrls) =>
        if (executorDataMap.contains(executorId)) {
          executorRef.send(RegisterExecutorFailed("Duplicate executor ID: " + executorId))
          context.reply(true)
        } else {
          // If the executor's rpc env is not listening for incoming connections, `hostPort`
          // will be null, and the client connection should be used to contact the executor.
          val executorAddress = if (executorRef.address != null) {
              executorRef.address
            } else {
              context.senderAddress
            }
          logInfo(s"Registered executor $executorRef ($executorAddress) with ID $executorId")
          addressToExecutorId(executorAddress) = executorId
          totalCoreCount.addAndGet(cores)
          totalRegisteredExecutors.addAndGet(1)
          val data = new ExecutorData(executorRef, executorRef.address, hostname,
            cores, cores, logUrls)
          // This must be synchronized because variables mutated
          // in this block are read when requesting executors
          CoarseGrainedSchedulerBackend.this.synchronized {
            executorDataMap.put(executorId, data)
            if (currentExecutorIdCounter < executorId.toInt) {
              currentExecutorIdCounter = executorId.toInt
            }
            if (numPendingExecutors > 0) {
              numPendingExecutors -= 1
              logDebug(s"Decremented number of pending executors ($numPendingExecutors left)")
            }
          }
          executorRef.send(RegisteredExecutor)
          // Note: some tests expect the reply to come after we put the executor in the map
          context.reply(true)
          listenerBus.post(
            SparkListenerExecutorAdded(System.currentTimeMillis(), executorId, data))
          makeOffers()
        }
......
......
}