通過WordCount詳解過程圖，觀察Spark的執行過程，思考RDD操作型別有幾種？有了初步的思考，下文會給出RDD操作型別區別和實戰演示

有如下三種：

（1）transformation：進行資料狀態的轉換，對已有的RDD建立新的RDD。

（2）Action：觸發具體的作業，對RDD最後取結果的一種操作

（3）Controller：對效能效率和容錯方面的支援。persist , cache, checkpoint

1. Transformation：進行資料狀態的轉換，對已有的RDD建立新的RDD。

2. Action：觸發具體的作業，對RDD最後取結果的一種操作。

（8）saveAsTextFile()

Spark action級別的操作，特點是內部都呼叫runJob方法，常用的action：reduce（1+2+3+4—-》1+2=3+3=6+4=10），aggregate，collection（把所有的結果放在driver上），count執行結果有多少條，take獲取執行結果的某幾條，saveAsTextFile把執行結果儲存在Hadoop支援的檔案系統上，foreach迴圈每一條元素；countByKey：可看一個key有多少個value；

注意：reduceByKey不是action，它是一個transformation

action級別操作舉例：

val numbers = sc.parallelize(1 to 100)
scala> numbers.reduce(_+_) 觸發作業
val result = numbers.map(2*_)
scala> result.collect
 

原始碼： 

 /**
   *Return an array that contains all of the elements in this RDD.
   */
  defcollect(): Array[T] = withScope {
    valresults = sc.runJob(this, (iter: Iterator[T]) => iter.toArray)
   Array.concat(results: _*)

  }

Driver會收集每個task的執行結果，每個Executor上會有很多的task。（通過網路通訊傳輸收集）

如果想在命令終端上看到執行結果就必須collect

凡是Action的級別的操作都會觸發sc.runJob。言外之意：一個應用程式會有很多的job。

scala> result.count()
def count(): Long = sc.runJob(this,Utils.getIteratorSize _).sum
scala> val topN = numbers.take(5)

countByKey：可看一個key有多少value

scala> val scores =Array(Tuple2(1,100),Tuple2(2,89),Tuple2(2, 99), Tuple2(3, 88))
scores: Array[(Int, Int)] = Array((1,100),(2,89), (2,99), (3,88))
scala> val scoresRDD =sc.parallelize(scores)
scoresRDD: org.apache.spark.rdd.RDD[(Int,Int)] = ParallelCollectionRDD[1] at parallelize at <console>:29
 defcountByKey(): Map[K, Long] = self.withScope {
   self.mapValues(_ => 1L).reduceByKey(_ + _).collect().toMap

  }

3. Controller：對效能效率和容錯方面的支援。persist , cache, checkpoint

並行度和Shuffle沒有任何關係，並行度是涉及到執行效率的，Shuffle由RDD的依賴關係決定，如上如果不指定並行度，並行度會傳遞過去的。設定多少個並行度就有多少個Partition。