spark（一）：spark概覽及邏輯執行圖

上圖是spark框架概要圖，spark一些重要概念先簡要介紹一下：

1. cluster manager：資源管理集群，比如standalone、yarn；
2. application：用戶編寫的應用程序；
3. Driver：application中的main函數，創建的SparkContext負責與cluster manager通信，進行資源的申請、任務的分配與監控。一般認為SparkContext就是Driver；
4. worker：集群中可以運行任務的節點；
5. executor：worker上運行任務的進程，負責執行task；
6. task：被executor執行的最小單位，一個stage由多個task組成；
7. stage：一個job中的多個階段，一般只要發生shuffle就會切分一個stage；
8. job：一個application至少有一個job，spark只要有一個action就會產生一個job。

spark邏輯執行圖的四個概要步驟

1. 從數據源創造初始RDD；
2. 對RDD進行一系列transformation操作，生成新的RDD[T]，其中類型T可以是scala中的基本數據類型，也可以是<k, v>，如果是<k, v>那么k不能是復雜數據結構；
3. 對最后的final RDD進行action操作，每個partition產生result；
4. 將result回送到Driver端，進行最后的計算。

邏輯執行圖的生成

1. 如何產生RDD，應該產生哪些RDD
   一般每個transformation方法都會返回一個RDD，有些transformation還會有一些子transformation，因此可能產生多于一個的RDD；
2. RDD的依賴關系
   RDD依賴哪些父RDD比較簡單，從代碼中可以直觀看到；
   RDD中有多少個partition呢？這個一般是用戶指定，如果未指定的話，會去父RDD中partition數最多的那個；
   RDD和父RDD的partitions之間是怎么依賴的呢？
   
   上圖前三種是窄依賴，最后一個是寬依賴。窄依賴一般也叫完全依賴，就是說父RDD中partition的全部數據都被子RDD特定的partition依賴；寬依賴一般也叫部分依賴，就是說父RDD中某個partition的一部分數據被子RDD的partition1所依賴，而另一部分數據被子RDD的partition2所依賴，這種情況就要發生shuflle。
   一般認為父RDD的所有partition只要不被子RDD的多個partition依賴就屬于窄依賴，就不會發生shuffle，但是存在特殊情況就是第三種情況：父RDD的partition被子RDD的多個partition依賴，但依然不需要發生shuffle（一般笛卡爾積是這種情況）。

常用transformation簡介

1. union：將兩個RDD合并，不改變partition里的數據
2. groupByKey：將相同key的records聚合在一起，聚合后的每條對應的value為原來所有相同的key的value組成的數組。（默認不會再map端開啟conbine）
3. reduceByKey：相當于傳統的MR，對相同key的value做出一定函數處理，得出最后一個value，比如reduceByKey(+)就會相同的key的value不斷相加。
   
   reduceByKey() 默認在 map 端開啟 combine()，因此在 shuffle 之前先通過 mapPartitions 操作進行 combine，得到 MapPartitionsRDD，然后 shuffle 得到 ShuffledRDD，然后再進行 reduce（通過 aggregate + mapPartitions() 操作來實現）得到 MapPartitionsRDD。
4. distinct：去重，這個transformation內部會先把value轉出<k, ->形式的rdd，然后進行依次reduceByKey，最后再還原。
5. cogroup(otherRdd, numPartitions)：類似groupByKey，不過這個聚合兩個或兩個以上的RDD，產生的結果也不太一樣，是每個RDD自己內部相同的key對應的value先聚合成一個數組，然后兩個rdd相同key對應的數組再聚合成一個二維數組，類似于[(a, c), (f)]這樣。
6. intersection(otherRdd)：抽取兩個rdd的公共數據，內部會想distinct那樣先把value轉為<k, ->形式，之后調用cogroup，最后把有相同key的留下并還原。
7. join(otherRdd)：將兩個 RDD[(K, V)] 按照 SQL 中的 join 方式聚合在一起。與 intersection() 類似，首先進行 cogroup()，得到<K, (Iterable[V1], Iterable[V2])>類型的 MappedValuesRDD，然后對 Iterable[V1] 和 Iterable[V2] 做笛卡爾集，并將集合 flat() 化。
8. sortByKey：將 RDD[(K, V)] 中的 records 按 key 排序，ascending = true 表示升序，false 表示降序。
9. cartesion：
   笛卡爾積就是上面提到的父RDD的partition被子RDD的多個partition依賴，但依然不需要發生shuffle的情況。
10. coalesce：當 shuffle = false 的時候，是不能增加 partition 個數的
11. filterByRange(lower: K, upper: K):以RDD中元素key的范圍做過濾，包含lower和upper上下邊界

spark常見action操作

1. reduce(func)：使用傳入的函數參數 func 對數據集中的元素進行匯聚操作 (兩兩合并).
2. collect()：在 driver program 上將數據集中的元素作為一個數組返回. 這在執行一個 filter 或是其他返回一個足夠小的子數據集操作后十分有用.
3. count()：返回數據集中的元素個數
4. first()：返回數據集中的第一個元素 (與 take(1) 類似)
5. take(n)：返回數據集中的前 n 個元素
6. takeOrdered(n, [ordering])：以其自然序或使用自定義的比較器返回 RDD 的前 n 元素
7. saveAsTextFile(path)：數據集中的元素寫入到指定目錄下的一個或多個文本文件中, 該目錄可以存在于本地文件系統, HDFS 或其他 Hadoop 支持的文件系統.
8. countByKey()：僅適用于 (K, V) 類型的 RDD. 返回每個 key 的 value 數的一個 hashmap (K, int) pair.
9. foreach(func)：對數據集中的每個元素執行函數 func.