RDD是什么？

RDD是Spark中的抽象數(shù)據(jù)結(jié)構(gòu)類型，任何數(shù)據(jù)在Spark中都被表示為RDD。從編程的角度來看，RDD可以簡單看成是一個(gè)數(shù)組。和普通數(shù)組的區(qū)別是，RDD中的數(shù)據(jù)是分區(qū)存儲(chǔ)的，這樣不同分區(qū)的數(shù)據(jù)就可以分布在不同的機(jī)器上，同時(shí)可以被并行處理。因此，Spark應(yīng)用程序所做的無非是把需要處理的數(shù)據(jù)轉(zhuǎn)換為RDD，然后對RDD進(jìn)行一系列的變換和操作從而得到結(jié)果。本文為第一部分，將介紹Spark RDD中與Map和Reduce相關(guān)的API中。

如何創(chuàng)建RDD？

RDD可以從普通數(shù)組創(chuàng)建出來，也可以從文件系統(tǒng)或者HDFS中的文件創(chuàng)建出來。

舉例：從普通數(shù)組創(chuàng)建RDD，里面包含了1到9這9個(gè)數(shù)字，它們分別在3個(gè)分區(qū)中。

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來創(chuàng)建RDD，文件中的每一行就是RDD中的一個(gè)元素

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

雖然還有別的方式可以創(chuàng)建RDD，但在本文中我們主要使用上述兩種方式來創(chuàng)建RDD以說明RDD的API。

map

map是對RDD中的每個(gè)元素都執(zhí)行一個(gè)指定的函數(shù)來產(chǎn)生一個(gè)新的RDD。任何原RDD中的元素在新RDD中都有且只有一個(gè)元素與之對應(yīng)。

舉例：

scala> val a = sc.parallelize(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中每個(gè)元素都乘以2來產(chǎn)生一個(gè)新的RDD。

mapPartitions

mapPartitions是map的一個(gè)變種。map的輸入函數(shù)是應(yīng)用于RDD中每個(gè)元素，而mapPartitions的輸入函數(shù)是應(yīng)用于每個(gè)分區(qū)，也就是把每個(gè)分區(qū)中的內(nèi)容作為整體來處理的。
它的函數(shù)定義為：

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

f即為輸入函數(shù)，它處理每個(gè)分區(qū)里面的內(nèi)容。每個(gè)分區(qū)中的內(nèi)容將以Iterator[T]傳遞給輸入函數(shù)f，f的輸出結(jié)果是Iterator[U]。最終的RDD由所有分區(qū)經(jīng)過輸入函數(shù)處理后的結(jié)果合并起來的。

舉例：

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))

上述例子中的函數(shù)myfunc是把分區(qū)中一個(gè)元素和它的下一個(gè)元素組成一個(gè)Tuple。因?yàn)榉謪^(qū)中最后一個(gè)元素沒有下一個(gè)元素了，所以(3,4)和(6,7)不在結(jié)果中。

mapPartitions還有些變種，比如mapPartitionsWithContext，它能把處理過程中的一些狀態(tài)信息傳遞給用戶指定的輸入函數(shù)。還有mapPartitionsWithIndex，它能把分區(qū)的index傳遞給用戶指定的輸入函數(shù)。

mapValues

mapValues顧名思義就是輸入函數(shù)應(yīng)用于RDD中Kev-Value的Value，原RDD中的Key保持不變，與新的Value一起組成新的RDD中的元素。因此，該函數(shù)只適用于元素為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,xlionx),(3,xcatx), (7,xpantherx), (5,xeaglex))

mapWith

mapWith是map的另外一個(gè)變種，map只需要一個(gè)輸入函數(shù)，而mapWith有兩個(gè)輸入函數(shù)。它的定義如下：

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

第一個(gè)函數(shù)constructA是把RDD的partition index（index從0開始）作為輸入，輸出為新類型A；

第二個(gè)函數(shù)f是把二元組(T, A)作為輸入（其中T為原RDD中的元素，A為第一個(gè)函數(shù)的輸出），輸出類型為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類似，區(qū)別是原RDD中的元素經(jīng)map處理后只能生成一個(gè)元素，而原RDD中的元素經(jīng)flatmap處理后可生成多個(gè)元素來構(gòu)建新RDD。

舉例：對原RDD中的每個(gè)元素x產(chǎn)生y個(gè)元素（從1到y(tǒng)，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很類似，都是接收兩個(gè)函數(shù)，一個(gè)函數(shù)把partitionIndex作為輸入，輸出是一個(gè)新類型A；另外一個(gè)函數(shù)是以二元組（T,A）作為輸入，輸出為一個(gè)序列，這些序列里面的元素組成了新的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應(yīng)用于元素為KV對的RDD中Value。每個(gè)一元素的Value被輸入函數(shù)映射為一系列的值，然后這些值再與原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中每個(gè)元素的值被轉(zhuǎn)換為一個(gè)序列（從其當(dāng)前值到5），比如第一個(gè)KV對(1,2), 其值2被轉(zhuǎn)換為2，3，4，5。然后其再與原KV對中Key組成一系列新的KV對(1,2),(1,3),(1,4),(1,5)。

reduce

reduce將RDD中元素兩兩傳遞給輸入函數(shù)，同時(shí)產(chǎn)生一個(gè)新的值，新產(chǎn)生的值與RDD中下一個(gè)元素再被傳遞給輸入函數(shù)直到最后只有一個(gè)值為止。

舉例

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進(jìn)行reduce，因此，Key相同的多個(gè)元素的值被reduce為一個(gè)值，然后與原RDD中的Key組成一個(gè)新的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的兩個(gè)元素被轉(zhuǎn)為了(3,10)。

Reference

本文中的部分例子來自:http://homepage.cs.latrobe.edu.au/zhe/ZhenHeSparkRDDAPIExamples.html

總結(jié)

以上就是本文關(guān)于淺談Spark RDD API中的Map和Reduce的全部內(nèi)容，希望對大家有所幫助。感興趣的朋友可以繼續(xù)參閱本站：Spark三種屬性配置方式詳解、淺談七種常見的Hadoop和Spark項(xiàng)目案例等，有什么問題可以隨時(shí)留言，小編會(huì)及時(shí)回復(fù)大家的。感謝朋友們對本站的支持！

最新評(píng)論