Spark RDD :Spark API--圖解Spark API

面試題引出：

簡述Spark的寬窄依賴，以及Spark如何劃分stage，每一個stage又根據什麼決定task個數? 

Stage：根據RDD之間的依賴關係的不一樣將Job劃分紅不一樣的Stage，遇到一個寬依賴則劃分一個Stage。

Task：Stage是一個TaskSet，將Stage根據分區數劃分紅一個個的Task。

 請列舉Spark的transformation算子（很多於8個），並簡述功能

1）map（func）：返回一個新的RDD，該RDD由每個輸入元素通過func函數轉換後組成.

2）mapPartitions(func)：相似於map，但獨立地在RDD的每個分片上運行，所以在類型爲T的RD上運行時，func的函數類型必須是Iterator[T] => Iterator[U]。假設有N個元素，有M個分區，那麼map的函數的將被調用N次,而mapPartitions被調用M次,一個函數一次處理全部分區。

3）reduceByKey（func，[numTask]）：在一個(K,V)的RDD上調用，返回一個(K,V)的RDD，使用定的reduce函數，將相同key的值聚合到一塊兒，reduce任務的個數能夠經過第二個可選的參數來設置。

4）aggregateByKey (zeroValue:U,[partitioner: Partitioner]) (seqOp: (U, V) => U,combOp: (U, U) => U: 在kv對的RDD中，，按key將value進行分組合並，合併時，將每一個value和初始值做爲seq函數的參數，進行計算，返回的結果做爲一個新的kv對，而後再將結果按照key進行合併，最後將每一個分組的value傳遞給combine函數進行計算（先將前兩個value進行計算，將返回結果和下一個value傳給combine函數，以此類推），將key與計算結果做爲一個新的kv對輸出。

 5）combineByKey(createCombiner: V=>C, mergeValue: (C, V) =>C, mergeCombiners: (C, C) =>C):

對相同K，把V合併成一個集合。

1.createCombiner: combineByKey() 會遍歷分區中的全部元素，所以每一個元素的鍵要麼尚未遇到過，要麼就和以前的某個元素的鍵相同。若是這是一個新的元素,combineByKey()會使用一個叫做createCombiner()的函數來建立那個鍵對應的累加器的初始值

2.mergeValue: 若是這是一個在處理當前分區以前已經遇到的鍵，它會使用mergeValue()方法將該鍵的累加器對應的當前值與這個新的值進行合併

3.mergeCombiners: 因爲每一個分區都是獨立處理的， 所以對於同一個鍵能夠有多個累加器。若是有兩個或者更多的分區都有對應同一個鍵的累加器， 就須要使用用戶提供的 mergeCombiners() 方法將各個分區的結果進行合併。

…

根據自身狀況選擇比較熟悉的算子加以介紹。

4.10.8 請列舉Spark的action算子（很多於6個），並簡述功能

1）reduce：

2）collect:

3）first：

4）take：

5）aggregate：

6）countByKey：

7）foreach：

8）saveAsTextFile：

4.10.9 請列舉會引發Shuffle過程的Spark算子，並簡述功能。

reduceBykey：

groupByKey：

…ByKey:

初識spark，須要對其API有熟悉的瞭解才能方便開發上層應用。本文用圖形的方式直觀表達相關API的工做特色，並提供瞭解新的API接口使用的方法。例子代碼所有使用python實現。

 

 

 

 

1. 數據源準備

準備輸入文件：

$ cat /tmp/in apple bag bag cat cat cat

啓動pyspark：

$ ./spark/bin/pyspark

使用textFile建立RDD:

>>> txt = sc.textFile("file:///tmp/in", 2)

查看RDD分區與數據：

>>> txt.glom().collect() [[u'apple', u'bag bag'], [u'cat cat cat']]

2. transformation

flatMap

處理RDD的每一行，一對多映射。

代碼示例：

>>> txt.flatMap(lambda line: line.split()).collect() [u'apple', u'bag', u'bag', u'cat', u'cat', u'cat']

示意圖：

 

 

 

map

處理RDD的每一行，一對一映射。

代碼示例：

>>> txt.flatMap(lambda line: line.split()).map(lambda word: (word, 1)).collect() [(u'apple', 1), (u'bag', 1), (u'bag', 1), (u'cat', 1), (u'cat', 1), (u'cat', 1)]

示意圖：

 

 

 

filter

處理RDD的每一行，過濾掉不知足條件的行。

代碼示例：

>>> txt.flatMap(lambda line: line.split()).filter(lambda word: word !='bag').collect() [u'apple', u'cat', u'cat', u'cat']

 

 

mapPartitions

逐個處理每個partition，使用迭代器it訪問每一個partition的行。

代碼示例：

>>> txt.flatMap(lambda line: line.split()).mapPartitions(lambda it: [len(list(it))]).collect() [3, 3]

示意圖：

 

 

mapPartitionsWithIndex

逐個處理每個partition，使用迭代器it訪問每一個partition的行，index保存partition的索引，等價於mapPartitionsWithSplit（過時函數）。

代碼示例：

>>> txt.flatMap(lambda line: line.split()).mapPartitionsWithIndex(lambda index, it: [index]).collect() [0, 1]

示意圖：

 

 

sample

根據採樣因子指定的比例，對數據進行採樣，能夠選擇是否用隨機數進行替換，seed用於指定隨機數生成器種子。第一個參數表示是否放回抽樣，第二個參數表示抽樣比例，第三個參數表示隨機數seed。

代碼示例：

>>> txt.flatMap(lambda line: line.split()).sample(False, 0.5, 5).collect() [u'bag', u'bag', u'cat', u'cat']

示意圖：

 

 

union

合併RDD，不去重。

代碼示例：

>>> txt.union(txt).collect() [u'apple', u'bag bag', u'cat cat cat', u'apple', u'bag bag', u'cat cat cat']

示意圖：

 

 

distinct

對RDD去重。

代碼示例：

>>> txt.flatMap(lambda line: line.split()).distinct().collect() [u'bag', u'apple', u'cat']

示意圖：

 

 

groupByKey

在一個（K,V）對的數據集上調用，返回一個（K，Seq[V])對的數據集。

代碼示例：

>>> txt.flatMap(lambda line: line.split()).map(lambda word: (word, 1)).groupByKey().collect() [(u'bag', <pyspark.resultiterable.ResultIterable object at 0x128a150>), (u'apple', <pyspark.resultiterable.ResultIterable object at 0x128a550>), (u'cat', <pyspark.resultiterable.ResultIterable object at 0x13234d0>)] >>> txt.flatMap(lambda line: line.split()).map(lambda word: (word, 1)).groupByKey().collect()[0][1].data [1, 1]

示意圖：

 

 

reduceByKey

在一個（K，V)對的數據集上調用時，返回一個（K，V）對的數據集，使用指定的reduce函數，將相同key的值聚合到一塊兒。

代碼示例：

>>> txt.flatMap(lambda line: line.split()).map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b).collect() [(u'bag', 2), (u'apple', 1), (u'cat', 3)]

示意圖：

 

 

aggregateByKey

自定義聚合函數，相似groupByKey。在一個（K,V）對的數據集上調用，不過能夠返回一個（K，Seq[U])對的數據集。

代碼示例(實現groupByKey的功能)：

>>> txt.flatMap(lambda line: line.split()).map(lambda word: (word, 1)).aggregateByKey([], lambda seq, elem: seq + [elem], lambda a, b: a + b).collect() [(u'bag', [1, 1]), (u'apple', [1]), (u'cat', [1, 1, 1])]

sortByKey

在一個（K，V)對的數據集上調用，K必須實現Ordered接口，返回一個按照Key進行排序的（K，V）對數據集。升序或降序由ascending布爾參數決定。

代碼示例：

>>> txt.flatMap(lambda line: line.split()).map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b).sortByKey().collect() [(u'apple', 1), (u'bag', 2), (u'cat', 3)]

示意圖：

 

 

join

在類型爲（K,V)和（K,W)類型的數據集上調用時，返回一個相同key對應的全部元素對在一塊兒的(K, (V, W))數據集。

代碼示例：

>>> sorted_txt = txt.flatMap(lambda line: line.split()).map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b).sortByKey() >>> sorted_txt.join(sorted_txt).collect() [(u'bag', (2, 2)), (u'apple', (1, 1)), (u'cat', (3, 3))]

示意圖：

 

 

cogroup

在類型爲（K,V)和（K,W)的數據集上調用，返回一個 (K, (Seq[V], Seq[W]))元組的數據集。這個操做也能夠稱之爲groupwith。

代碼示例：

>>> sorted_txt = txt.flatMap(lambda line: line.split()).map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b).sortByKey() >>> sorted_txt.cogroup(sorted_txt).collect() [(u'bag', (<pyspark.resultiterable.ResultIterable object at 0x1323790>, <pyspark.resultiterable.ResultIterable object at 0x1323310>)), (u'apple', (<pyspark.resultiterable.ResultIterable object at 0x1323990>, <pyspark.resultiterable.ResultIterable object at 0x1323ad0>)), (u'cat', (<pyspark.resultiterable.ResultIterable object at 0x1323110>, <pyspark.resultiterable.ResultIterable object at 0x13230d0>))] >>> sorted_txt.cogroup(sorted_txt).collect()[0][1][0].data [2]

示意圖：

 

cartesian

笛卡爾積，在類型爲 T 和 U 類型的數據集上調用時，返回一個 (T, U)對數據集(兩兩的元素對)。

代碼示例：

>>> sorted_txt = txt.flatMap(lambda line: line.split()).map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b).sortByKey() >>> sorted_txt.cogroup(sorted_txt).collect() [(u'bag', (<pyspark.resultiterable.ResultIterable object at 0x1323790>, <pyspark.resultiterable.ResultIterable object at 0x1323310>)), (u'apple', (<pyspark.resultiterable.ResultIterable object at 0x1323990>, <pyspark.resultiterable.ResultIterable object at 0x1323ad0>)), (u'cat', (<pyspark.resultiterable.ResultIterable object at 0x1323110>, <pyspark.resultiterable.ResultIterable object at 0x13230d0>))] >>> sorted_txt.cogroup(sorted_txt).collect()[0][1][0].data [2]

示意圖：

 

 

 

 

pipe

處理RDD的每一行做爲shell命令輸入，shell命令結果爲輸出。

代碼示例：

>>> txt.pipe("awk '{print $1}'").collect() [u'apple', u'bag', u'cat']

示意圖：

 

 

coalesce

減小RDD分區數。

代碼示例：

>>> txt.coalesce(1).collect() [u'apple', u'bag bag', u'cat cat cat']

示意圖：

 

 

repartition

對RDD從新分區，相似於coalesce。

代碼示例：

>>> txt.repartition(1).collect() [u'apple', u'bag bag', u'cat cat cat']

zip

合併兩個RDD序列爲元組，要求序列長度相等。

代碼示例：

>>> txt.zip(txt).collect() [(u'apple', u'apple'), (u'bag bag', u'bag bag'), (u'cat cat cat', u'cat cat cat')]

示意圖：

 

 

3. action

reduce

彙集數據集中的全部元素。

代碼示例：

>>> txt.reduce(lambda a, b: a + " " + b) u'apple bag bag cat cat cat'

示意圖：

 

 

collect

以數組的形式，返回數據集的全部元素。

代碼示例：

>>> txt.collect() [u'apple', u'bag bag', u'cat cat cat']

count

返回數據集的元素的個數。

代碼示例：

>>> txt.count() 3

first

返回數據集第一個元素。

代碼示例：

>>> txt.first() u'apple'

take

返回數據集前n個元素。

代碼示例：

>>> txt.take(2) [u'apple', u'bag bag']

takeSample

採樣返回數據集前n個元素。第一個參數表示是否放回抽樣，第二個參數表示抽樣個數，第三個參數表示隨機數seed。

代碼示例：

>>> txt.takeSample(False, 2, 1) [u'cat cat cat', u'bag bag']

takeOrdered

排序返回前n個元素。

代碼示例：

>>> txt.takeOrdered(2) [u'apple', u'bag bag']

saveAsTextFile

將數據集的元素，以textfile的形式，保存到本地文件系統，HDFS或者任何其它hadoop支持的文件系統。

代碼示例：

>>> txt.flatMap(lambda line: line.split(" ")).map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b).saveAsTextFile("file:///tmp/out")

查看輸出文件：

$cat /tmp/out/part-00001 (u'bag', 2) (u'apple', 1) (u'cat', 3)

saveAsSequenceFile

將數據集的元素，以Hadoop sequencefile的格式，保存到指定的目錄下，本地系統，HDFS或者任何其它hadoop支持的文件系統。這個只限於由key-value對組成，並實現了Hadoop的Writable接口，或者隱式的能夠轉換爲Writable的RDD。

countByKey

對(K,V)類型的RDD有效，返回一個(K，Int)對的Map，表示每個key對應的元素個數。

代碼示例：

>>> txt.flatMap(lambda line: line.split(" ")).map(lambda word: (word, 1)).countByKey() defaultdict(<type 'int'>, {u'bag': 2, u'apple': 1, u'cat': 3})

foreach

在數據集的每個元素上，運行函數func進行更新。這一般用於邊緣效果，例如更新一個累加器，或者和外部存儲系統進行交互。

代碼示例：

>>> def func(line): print line >>> txt.foreach(lambda line: func(line)) apple bag bag cat cat cat