Spark RDD API詳解之：Map和Reduce

RDD是什麼？

RDD是Spark中的抽象數據結構類型，任何數據在Spark中都被表示爲RDD。從編程的角度來看，

RDD能夠簡單當作是一個數組。和普通數組的區別是，RDD中的數據是分區存儲的，這樣不一樣分區的數據就能夠分佈在不一樣的機器上，同時能夠被並行處理。所以,spark應用程序所作的無非是把須要處理的數據轉換成RDD，而後對RDD進行一系列的變換和操做從而獲得結果。本文爲第一部分，將介紹Spark RDD中與Map和Reduce相關的API。

如何建立RDD?

RDD能夠從普通數組建立出來，也能夠從文件系統或者HDFS中的文件建立出來。

方式(一)：舉例：從普通數組建立RDD,裏面包含了1到9這9個數字，他們分別在3個分區中。

scala>val a=sc.parallelize(1 to 9, 3)
a:org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at parallelize at <console>:12

方式(二):舉例：讀取文件README.md來建立RDD,文件中的每一行就是RDD中的一個元素

scala> val b = sc.textFile("README.md")
b: org.apache.spark.rdd.RDD[String] = MappedRDD[3] at textFile at <console>:12

Text file RDDs的建立可使用SparkContext的textFile方法。該方法接受一個文件的URI地址(或者是機器上的一個本地路徑，或者是一個hdfs://，等URI)做爲參數，並讀取文件的一行數據，放入集合中。下面是一個調用例子：

scala>val distFile = sc.textFile("data.txt")
distFile:RDD[String]= MappedRDD@1D4CEE08

一旦建立完成，就能夠在distFile上執行數據集操做。例如：想要對全部行的長度進行求和，咱們就能夠經過以下的map和reduce操做來完成：

distFile.map(s =>s.length).reduce((a+b) =>a+b)

雖然還有別的方式能夠建立RDD,但在本文中咱們主要使用上述兩種方式來建立RDD以說明RDD的API.

map

map是對RDD中的每一個元素都執行一個指定的函數來產生一個新的RDD。任何原RDD中的元素在新RDD中都有且只有一個元素與之對應。

舉例：

scala>val a = sc.paralleize(1 to 9, 3)
scala>val b = a.map(x=>x*2)
scala>a.collect
res10:Array[Int] = Array(1,2,3,4,5,6,7,8,9)
scala>b.collect
res11:Array[Int] = Array(2,4,6,8,10,12,14,16,18)

上述例子中把原RDD中每一個元素都乘以2來產生一個新的RDD。

mapPartitions

mapPartitions是map的一個變種。map的輸入函數是應用於RDD中每一個元素，而mapPartitions的輸入函數是應用於每一個分區，也就是把每一個分區中的內容做爲總體來處理的。

它的函數定義爲：

def mapPartitions[U:ClassTag](f:Iterator[T] =>Iterator[U],preserversPartitioning:Boolean=false:RDD[U])

f即爲輸入函數，它處理每一個分區裏面的內容。每一個分區中的內容將以Iterator[T]傳遞給輸入函數f，f的輸出結果是Iterator[U]。最終的RDD由全部分區通過輸入函數處理後的結果合併起來的。

舉例：

scala>val a = sc.parallelize(1 to 9,3)
scala>def myfunc[T](iter:Iterator[T]:Iterator[(T,T)] = {
   var  res = List[(T,T)]()
   var  pre = iter.next.while(iter,hasNext){
    val cur = iter.next;
    res.::=(pre,cur) pre = cur;
} 
   res.iterator


}) 
scala>a.mapPartitions(myfunc).collect
res0:Array[(Int,Int)] = Array((2,3),(1,2),(5,6),(4,5),(8,9),(7.8))

上述例子中的函數myfunc是把分區中一個元素和它的下一個元素組成一個Tuple。由於分區中最後一個元素沒有下一個元素了，因此(3,4)和(6,7)不在結果中。 
mapPartitions還有些變種，好比mapPartitionsWithContext，它能把處理過程當中的一些狀態信息傳遞給用戶指定的輸入函數。還有mapPartitionsWithIndex，它能把分區的index傳遞給用戶指定的輸入函數。

mapValuesmapValues顧名思義就是輸入函數應用於RDD中key-value的Value。原RDD中的key保持不變，與新的value一塊兒組成新的RDD中的元素。所以，該函數只適用於元素爲KV對的RDD.

scala>val a = sc.parallelize(List("dog","tiger","lion","cat","panther","eagle"),2)
scala>val b =a.map(x=>(x.length,x))
scala>b.mapValues("x"+_+"x").collect
res5:Array[(Int,String)] = Array((3,xdogx),(5,xtigerx),(4,xlions),(3,xcatx),(7,xpantherx),(5,xeaglex))

mapWith

mapWth是map的另一個變種，map只須要一個輸入函數,而mapWth有兩個輸入函數。它的定義以下：def mapWith[A: ClassTag, U: ](constructA: Int => A, preservesPartitioning: Boolean = false)(f: (T, A) => U): RDD[U]

• 第一個函數constructA是把RDD的partition index（index從0開始）做爲輸入，輸出爲新類型A；
• 第二個函數f是把二元組(T, A)做爲輸入（其中T爲原RDD中的元素，A爲第一個函數的輸出），輸出類型爲U。

舉例：把partition index乘以10，而後加上2做爲新的RDD的元素。

val x = sc.parallelize(List(1,2,3,4,5,6,7,8,9,10),3)
x.mapWith(a=>a*10)((a,b)=>(b+2)).collect
res4:Array[Int] = Array(2,2,2,12,12,12,22,22,22,22)

flatMap

與map相似，區別是原RDD中的元素經map處理後只能生產一個元素，而原RDD中的元素經flatmap處理後可生成多個元素來構建新RDD。 
舉例：對原RDD中的每一個元素x產生y個元素（從1到y，y爲元素x的值）

scala>val a = sc.parallelize(1 to 4,2)
scala>val b = a.flatMap(x=>1 to x)
scala>b.collect
res12:Array[Int] =Array(1,1,2,1,2,3,1,2,3,4)

flatMapWith

flatMapWith與mapWith很相似，都是接收兩個函數，一個函數把partitionIndex做爲輸入，輸出是一個新類型A；另一個函數是以二元組（T,A）做爲輸入，輸出爲一個序列，這些序列裏面的元素組成了新的RDD。它的定義以下：

def flatMapWith[A: ClassTag, U: ClassTag](constructA: Int => A, preservesPartitioning: Boolean = false)(f: (T, A) => Seq[U]): RDD[U]

舉例：

scala> val a = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 3)
scala> a.flatMapWith(x => x, true)((x, y) => List(y, x)).collect
res58: Array[Int] = Array(0, 1, 0, 2, 0, 3, 1, 4, 1, 5, 1, 6, 2, 7, 2,
8, 2, 9)

flatMapValues

flatMapValues相似於mapValues，不一樣的在於flatMapValues應用於元素爲KV對的RDD中Value。每一個一元素的Value被輸入函數映射爲一系列的值，而後這些值再與原RDD中的Key組成一系列新的KV對。

舉例

scala> val a = sc.parallelize(List((1,2),(3,4),(3,6)))
scala> val b = a.flatMapValues(x=>x.to(5))
scala> b.collect
res3: Array[(Int, Int)] = Array((1,2), (1,3), (1,4), (1,5), (3,4), (3,5))

上述例子中原RDD中每一個元素的值被轉換爲一個序列（從其當前值到5），好比第一個KV對(1,2), 其值2被轉換爲2，3，4，5。而後其再與原KV對中Key組成一系列新的KV對(1,2),(1,3),(1,4),(1,5)。

reduce

reduce將RDD中元素兩兩傳遞給輸入函數，同時產生一個新的值，新產生的值與RDD中下一個元素再被傳遞給輸入函數直到最後只有一個值爲止。

舉例

scala> val c = sc.parallelize(1 to 10)
scala> c.reduce((x, y) => x + y)
res4: Int = 55

上述例子中對RDD中的元素求和。

reduceByKey

顧名思義，reduceByKey就是對元素爲KV對的RDD中Key相同的元素的Value進行reduce，所以，Key相同的多個元素的值被reduce爲一個值，而後與原RDD中的Key組成一個新的KV對。

舉例:

scala> val a = sc.parallelize(List((1,2),(3,4),(3,6)))
scala> a.reduceByKey((x,y) => x + y).collect
res7: Array[(Int, Int)] = Array((1,2), (3,10))

上述例子中，對Key相同的元素的值求和，所以Key爲3的兩個元素被轉爲了(3,10)。