[Apache Spark源码阅读]天堂之门——SparkContext解析

稍微了解Spark源码的人应该都知道SparkContext，作为整个Project的程序入口，其重要性不言而喻，许多大牛也在源码分析的文章中对其做了很多相关的深入分析和解读。这里，结合自己前段时间的阅读体会，与大家共同讨论学习一下Spark的入口对象—天堂之门—SparkContex。

SparkContex位于项目的源码路径spark-mastercoresrcmainscalaorgapachesparkSparkContext.scala中，源文件包含Classs SparkContext声明和其伴生对象object SparkContext。而之所以将SparkContext称为整个程序的入口，原因在于，不管我们是从本地还是HDFS读取文件，总要首先创建一个SparkContext对象，然后基于这个SC对象，展开后续的RDD对象创建、转换等操作。

在创建SparkContex对象的过程中，进行了一系列的初始化操作，主要包括以下内容：

1. 载入配置文件SparkConf
2. 创建SparkEnv
3. 创建TaskScheduler
4. 创建DAGScheduler

1、 载入配置文件SparkConf

在SparkConf初始化时，会将相关的配置参数传递给SparkContex，包括master、appName、sparkHome、jars、environment等信息，这里的构造函数有多中表达形式，但最归初始化的结果都是殊途同归，SparkContex获取了所有相关的本地配置和运行时配置信息。

def this() = this(new SparkConf())

def this(master: String, appName: String, conf: SparkConf) =
    this(SparkContext.updatedConf(conf, master, appName))

def this(
      master: String,
      appName: String,
      sparkHome: String = null,
      jars: Seq[String] = Nil,
      environment: Map[String, String] = Map(),
      preferredNodeLocationData: Map[String, Set[SplitInfo]] = Map()) =
  {
    this(SparkContext.updatedConf(new SparkConf(), master, appName, sparkHome, jars, environment))
    this.preferredNodeLocationData = preferredNodeLocationData
  }

 

2、创建SparkEnv

SparkEnv是一个非常重要的变量，其内包含了许多Spark运行时的重要组件（变量），包括 MapOutputTracker、ShuffleFetcher、BlockManager等，这里是通过SparkEnv类的伴生对象SparkEnv Object内的Create方法实现的。

  private[spark] val env = SparkEnv.create(
    conf,
    "<driver>",
    conf.get("spark.driver.host"),
    conf.get("spark.driver.port").toInt,
    isDriver = true,
    isLocal = isLocal,
    listenerBus = listenerBus)
  SparkEnv.set(env)

 

3、创建TaskScheduler和DAGScheduler

下面这段代码非常重要，它初始化了SparkContex里两个非常关键的变量，TaskScheduler和DAGScheduler。

  private[spark] var taskScheduler = SparkContext.createTaskScheduler(this, master)
  @volatile private[spark] var dagScheduler: DAGScheduler = _
  try {
    dagScheduler = new DAGScheduler(this)
  } catch {
    case e: Exception => throw
      new SparkException("DAGScheduler cannot be initialized due to %s".format(e.getMessage))
  }

  // start TaskScheduler after taskScheduler sets DAGScheduler reference in DAGScheduler's
  // constructor
  taskScheduler.start()

 

首先，TaskScheduler是根据Spark的运行模式进行初始化的，具体代码在SparkContext中的createTaskScheduler方法中。以Standalone模式为例，它会将sc传递给TaskSchedulerImpl，并在返回Scheduler对象之前，创建SparkDeploySchedulerBackend，并将其初始化，最后返回Scheduler对象。

    case SPARK_REGEX(sparkUrl) =>
        val scheduler = new TaskSchedulerImpl(sc)
        val masterUrls = sparkUrl.split(",").map("spark://" + _)
        val backend = new SparkDeploySchedulerBackend(scheduler, sc, masterUrls)
        scheduler.initialize(backend)
        scheduler

创建TaskScheduler对象后，再将TaskScheduler对象传参至DAGScheduler，用来创建DAGScheduler对象，

  def this(sc: SparkContext, taskScheduler: TaskScheduler) = {
    this(
      sc,
      taskScheduler,
      sc.listenerBus,
      sc.env.mapOutputTracker.asInstanceOf[MapOutputTrackerMaster],
      sc.env.blockManager.master,
      sc.env)
  }

之后，再调用其start()方法将其启动，其中包括SchedulerBackend的启动。

override def start() {
    backend.start()

    if (!isLocal && conf.getBoolean("spark.speculation", false)) {
      logInfo("Starting speculative execution thread")
      import sc.env.actorSystem.dispatcher
      sc.env.actorSystem.scheduler.schedule(SPECULATION_INTERVAL milliseconds,
            SPECULATION_INTERVAL milliseconds) {
        Utils.tryOrExit { checkSpeculatableTasks() }
      }
    }
  }

 

除此之外，SparkContex还包括一些重要的函数方法，例如

1、runjob

runjob是spark中所有任务提交的入口，诸如rdd中的一些常见操作和变换，都会调用SparkContex的runjob方法，提交任务。

  def runJob[T, U: ClassTag](
      rdd: RDD[T],
      func: (TaskContext, Iterator[T]) => U,
      partitions: Seq[Int],
      allowLocal: Boolean,
      resultHandler: (Int, U) => Unit) {
    if (dagScheduler == null) {
      throw new SparkException("SparkContext has been shutdown")
    }
    val callSite = getCallSite
    val cleanedFunc = clean(func)
    logInfo("Starting job: " + callSite)
    val start = System.nanoTime
    dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, allowLocal,
      resultHandler, localProperties.get)
    logInfo("Job finished: " + callSite + ", took " + (System.nanoTime - start) / 1e9 + " s")
    rdd.doCheckpoint()
  }

2、textFile

从HDFS路径读取单个数据文件，首先创建HadoopRDD，通过map操作，返回RDD对象。

3、wholeTextFiles

从HDFS某个文件夹读取多个文件。

4、parallelize

读取本地文件，并转换为RDD。