SparkContext是通往Spark集群的唯一入口，是整個Application運行調度的核心。

一、Spark Driver Program

　　Spark Driver Program（以下簡稱Driver）是運行Application的main函數並且新建SparkContext實例的程式。其實，初始化SparkContext是為了準備Spark應用程式的運行環境，在Spark中，由SparkContext負責與集群進行通訊、資源的申請、任務的分配和監控等。當Worker節點中的Executor運行完畢Task後，Driver同時負責將SparkContext關閉。通常也可以使用SparkContext來代表驅動程式（Driver）。

　　Driver（SparkContext）整體架構圖如下所示。

二、SparkContext深度剖析

　　SparkContext是通往Spark集群的唯一入口，可以用來在Spark集群中創建RDDs、累加器（Accumulators）和廣播變數（Broadcast Variables）。SparkContext也是整個Spark應用程式（Application）中至關重要的一個對象，可以說是整個Application運行調度的核心（不是指資源調度）。

　　SparkContext的核心作用是初始化Spark應用程式運行所需要的核心組件，包括高層調度器（DAGScheduler）、底層調度器（TaskScheduler）和調度器的通訊終端（SchedulerBackend），同時還會負責Spark程式向Master註冊程式等。

　　一般而言，通常為了測試或者學習Spark開發一個Application，在Application的main方法中，最開始幾行編寫的程式碼一般是這樣的：

　　首先，創建SparkConf實例，設置SparkConf實例的屬性，以便覆蓋Spark默認配置文件spark-env.sh,spark-default.sh和log4j.properties中的參數；

　　然後，SparkConf實例作為SparkContext類的唯一構造參數來實例化SparkContext實例對象。SparkContext在實例化的過程中會初始化DAGScheduler、TaskScheduler和SchedulerBackend，而當RDD的action觸發了作業（Job）後，SparkContext會調用DAGScheduler將整個Job劃分成幾個小的階段（Stage），TaskScheduler會調度每個Stage的任務（Task）進行處理。

　　還有，SchedulerBackend管理整個集群中為這個當前的Application分配的計算資源，即Executor。

　　如果用一個車來比喻Spark Application，那麼SparkContext就是車的引擎，而SparkConf是關於引擎的配置參數。說明：只可以有一個SparkContext實例運行在一個JVM記憶體中，所以在創建新的SparkContext實例前，必須調用stop方法停止當前JVM唯一運行的SparkContext實例。

　　Spark程式在運行時分為Driver和Executor兩部分：Spark程式編寫是基於SparkContext的，具體包含兩方面。

　　Spark編程的核心基礎RDD是由SparkContext最初創建的（第一個RDD一定是由SparkContext創建的）。

　　Spark程式的調度優化也是基於SparkContext，首先進行調度優化。

　　Spark程式的註冊是通過SparkContext實例化時生產的對象來完成的（其實是SchedulerBackend來註冊程式）。

　　Spark程式在運行時要通過Cluster Manager獲取具體的計算資源，計算資源獲取也是通過SparkContext產生的對象來申請的（其實是SchedulerBackend來獲取計算資源的）。

　　SparkContext崩潰或者結束的時候，整個Spark程式也結束。

三、SparkContext源碼解析

　　SparkContext是Spark應用程式的核心。我們運行WordCount程式，通過日誌來深入了解SparkContext。

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.rdd.HadoopRDD
import org.apache.log4j.Logger
import org.apache.log4j.Level

object wordcount {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ALL);
    
    // 第1步：創建Spark的配置對象SparkConf，設置Spark程式運行時的配置資訊，
    val conf = new SparkConf().setAppName("My First Spark APP").setMaster("local")
    
    // 第2步：創建SparkContext對象
    val sc = new SparkContext(conf)
    
    // 第3步：根據具體的數據來源來創建RDD
    val lines = sc.textFile("helloSpark.txt", 1)
    
    // 第4步：對初始的RDD進行Transformation級別的處理，如通過map、filter等
    val words = lines.flatMap{line=>line.split(" ")}
    val pairs = words.map{word=>(word,1)}
    pairs.cache()
    val wordCountsOdered = pairs.reduceByKey(_+_).saveAsTextFile("wordCountResult.log")
//    val wordCountsOdered = pairs.reduceByKey(_+_).map(
//      pair=>(pair._2,pair._1)    
//    ).sortByKey(false).map(pair=>(pair._2,pair._1))
//    wordCountsOdered.collect.foreach(wordNumberPair=>println(wordNumberPair._1+" : "+wordNumberPair._2))
//    while(true){
//      
//    }
    sc.stop()
    
    
  }
}

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

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

    // 第1步：創建Spark的配置對象SparkConf，設置Spark程式運行時的配置資訊，
    val conf = new SparkConf().setAppName("My First Spark APP").setMaster("local")
    
    // 第2步：創建SparkContext對象
    val sc = new SparkContext(conf)

/**
 * A collection of regexes for extracting information from the master string.
 */
private object SparkMasterRegex {
  // Regular expression used for local[N] and local[*] master formats
  val LOCAL_N_REGEX = """local\[([0-9]+|\*)\]""".r
  // Regular expression for local[N, maxRetries], used in tests with failing tasks
  val LOCAL_N_FAILURES_REGEX = """local\[([0-9]+|\*)\s*,\s*([0-9]+)\]""".r
  // Regular expression for simulating a Spark cluster of [N, cores, memory] locally
  val LOCAL_CLUSTER_REGEX = """local-cluster\[\s*([0-9]+)\s*,\s*([0-9]+)\s*,\s*([0-9]+)\s*]""".r
  // Regular expression for connecting to Spark deploy clusters
  val SPARK_REGEX = """spark://(.*)""".r
  // Regular expression for connection to Mesos cluster by mesos:// or mesos://zk:// url
  val MESOS_REGEX = """mesos://(.*)""".r
}

  /**
   * 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 SPARK_REGEX(sparkUrl) =>
        val scheduler = new TaskSchedulerImpl(sc)
        val masterUrls = sparkUrl.split(",").map("spark://" + _)
        val backend = new StandaloneSchedulerBackend(scheduler, sc, masterUrls)
        scheduler.initialize(backend)
        (backend, scheduler)

  def start() {
    // Just launch an rpcEndpoint; it will call back into the listener.
    endpoint.set(rpcEnv.setupEndpoint("AppClient", new ClientEndpoint(rpcEnv)))
  }