以wordcount流程解析

1. val lines = sc.textFile()

  def textFile(
      path: String,
      minPartitions: Int = defaultMinPartitions): RDD[String] = withScope {
    assertNotStopped()
    //hadoopFile（）方法的呼叫，拿到Hadoop的配置檔案，建立HadoopRDD，廣播變數
    hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text],
    //執行map運算元操作，剔除key，只保留Value，獲得一個MapPartionsRDD。
    //MapPartionsRDD裡面就是一行一行的文字資料 
      minPartitions).map(pair => pair._2.toString).setName(path)
  }
 

1. val words = lines.flatMap(line => line.split(” “)) val pairs =

2. words.map(word => (word, 1))

// 其實RDD裡是沒有reduceByKey的，因此對RDD呼叫reduceByKey()方法的時候，會觸發scala的隱式轉換；此時就會在作用域內，尋找隱式轉換，會在RDD中找到rddToPairRDDFunctions()隱式轉換，然後將RDD轉換為PairRDDFunctions。
// 接著會呼叫PairRDDFunctions中的reduceByKey()方法

1. val counts = pairs.reduceByKey(_ + _)

2. counts.foreach(count => println(count._1 + “: ” + count._2))

  def runJob[T, U: ClassTag](
      rdd: RDD[T],
      func: (TaskContext, Iterator[T]) => U,
      partitions: Seq[Int],
      resultHandler: (Int, U) => Unit): Unit = {
    if (stopped.get()) {
      throw new IllegalStateException("SparkContext has been shutdown"
 
)
    }
    val callSite = getCallSite
    val cleanedFunc = clean(func)
    logInfo("Starting job: " + callSite.shortForm)
    if (conf.getBoolean("spark.logLineage", false)) {
      logInfo("RDD's recursive dependencies:\n" + rdd.toDebugString)
    }
    //呼叫SparkContext之前初始化建立的DAGScheduler的Runjob的方法。
    dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, resultHandler, localProperties.get)
    progressBar.foreach(_.finishAll())
    rdd.doCheckpoint()
  }