RDD提供了一組非常豐富的操作來操作數(shù)據(jù)，如：map,flatMap,filter等轉(zhuǎn)換操作，以及SaveAsTextFile,conutByKey等行動操作。這里僅僅綜述了轉(zhuǎn)換操作。

• map

map是對RDD中的每一個元素都執(zhí)行一個指定的函數(shù)來產(chǎn)生一個新的RDD，RDD之間的元素是一對一的關(guān)系。

val rdd1: RDD[Int] = sc.parallelize(1 to 9, 3) val rdd2: RDD[Int] = rdd1.map(_ * 2) printResult("map", rdd2) // 結(jié)果：map >> List(2, 4, 6, 8, 10, 12, 14, 16, 18)

• flapMap

flatMap類似于map，但是每一個輸入元素，會被映射為0到多個出輸出元素(即func函數(shù)的返回值是一個Seq,而不是單一元素)的新的RDD，RDD之間的元素是一對多關(guān)系。

val rdd3: RDD[Int] = rdd2.filter(x => x > 10).flatMap(x => x to 21) printResult("flatMap", rdd3) // 結(jié)果：flatMap >> List(12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 14, 15, 16, 17, 18, 19, 20, 21, 16, 17, 18, 19, 20, 21, 18, 19, 20, 21)

• filter

filter是對RDD元素進(jìn)行過濾，返回一個新的數(shù)據(jù)集，有經(jīng)過func函數(shù)后返回值為true的元素組成。

val rdd4 = rdd2.filter(x => x > 11) printResult("filter", rdd4) // 結(jié)果：filter >> List(12, 14, 16, 18)

• mapPartitions

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

函數(shù)定義：

def mapPartitions[U: ClassTag](f: Iterator[T] => Iterator[U],preservesPartitioning: Boolean = false): RDD[U] val rdd5: RDD[(Int, Int)] = rdd1.repartition(2).mapPartitions((iter: Iterator[Int]) => { val lst: ListBuffer[(Int, Int)] = new ListBuffer[(Int, Int)]() var prev: Int = 0 var current: Int = 0while (iter.hasNext) {current = iter.nextlst += ((prev, current))prev = current}lst.iterator}) printResult("mapPartitions", rdd5) 結(jié)果：mapPartitions >> List((0,1), (1,3), (3,5), (5,7), (7,9), (0,2), (2,4), (4,6), (6,8))

• mapPartitionsWithIndex

mapPartitionsWithIndex于mapPartitions的功能類似，只是多傳入split index而已，所有func函數(shù)必須是(Int,Iterator<T>)=> Iterator<U>類型。

val rdd6 = rdd1.repartition(2).mapPartitionsWithIndex((idx, iter) => { val lst = new ListBuffer[String]() var sum = 0 while (iter.hasNext) {sum += iter.next}lst += (idx + ":" + sum)lst.iterator }) printResult("mapPartitionsWithIndex", rdd6) // 結(jié)果：mapPartitionsWithIndex >> List(0:25, 1:20)

• sample

sample(withReplacemet,fraction,seed)是根據(jù)給定的隨機種子seed,隨機抽樣出數(shù)量為fraction的數(shù)據(jù)。其中，withReplacement：是否放回抽樣；fraction：比抽樣比例，0.1表示抽樣10%，seed：隨機種子，相同的seed得到的隨機序列是一樣的。所以，如果沒有設(shè)置seed,同一段代碼執(zhí)行兩遍得到的隨機序列是一樣的。
?

val sampleRDD = sc.parallelize(0 to 1000, 2) val rdd7 = sampleRDD.sample(false, 0.02, 2) printResult("sample", rdd7) // 結(jié)果：sample >> List(10, 42, 110, 121, 145, 158, 166, 234, 237, 253, 266, 343, 354, 393, 404, 457, 460, 662, 728, 738, 806, 808, 868, 887, 889, 934, 952)

• union

union(otherDataset)是數(shù)據(jù)的合并，返回一個新的數(shù)據(jù)集，由原數(shù)據(jù)集和otherDataset合并而成的一個數(shù)據(jù)集RDD。

val rdd8 = rdd1.union(rdd2) printResult("union", rdd8) // 結(jié)果：union >> List(1, 2, 3, 4, 5, 6, 7, 8, 9, 2, 4, 6, 8, 10, 12, 14, 16, 18)

• intersection

intersection(otherDataset)是數(shù)據(jù)交集，返回一個新的數(shù)據(jù)集，它是兩個數(shù)據(jù)集的交集數(shù)據(jù)。

val rdd9 = rdd8.intersection(rdd1) printResult("intersection", rdd9) // 結(jié)果：intersection >> List(4, 8, 1, 9, 5, 6, 2, 3, 7)

• distinct

distince(numPartitions)是對數(shù)據(jù)集進(jìn)行去重，返回一個新的數(shù)據(jù)集，它是對兩個數(shù)據(jù)集去掉重復(fù)數(shù)據(jù)后得到的一個數(shù)據(jù)集。其中，numPartitions參數(shù)是設(shè)置任務(wù)并行數(shù)量。

val rdd10 = rdd8.union(rdd9).distinct(2) printResult("distinct", rdd10) // 結(jié)果：distinct >> List(4, 16, 14, 6, 8, 12, 18, 10, 2, 1, 3, 7, 9, 5)

• groupByKey

groupByKey(partitioner)是對數(shù)據(jù)進(jìn)行分組操作，在一個由(K,V)對組成的數(shù)據(jù)集上調(diào)用，返回一個(K,Seq[V])對的數(shù)據(jù)集。注意，默認(rèn)情況下，使用8個并行任務(wù)進(jìn)行分組，可以傳入partitioner參數(shù)設(shè)置并行任務(wù)的分區(qū)數(shù).

val rddMap: RDD[(Int, Int)] = rdd1.map(item => (item % 3, item)) val rdd11 = rddMap.groupByKey(); printResult("groupByKey", rdd11) // 結(jié)果：groupByKey >> List((0,CompactBuffer(3, 6, 9)), (2,CompactBuffer(2, 5, 8)), (1,CompactBuffer(1, 4, 7)))

• reduceByKey

reduceByKey(func, numPartitions)是對數(shù)據(jù)進(jìn)行分組聚合操作，在一個(K,V)對的數(shù)據(jù)集上使用，返回一個(K,V)對的數(shù)據(jù)集。Key是相同的值，都被使用指定的reduce函數(shù)聚合到一起。和groupByKey類似，可以通過參數(shù)numPartitions設(shè)置并行任務(wù)的分區(qū)數(shù)。

val rdd12 = rddMap.reduceByKey((x, y) => x + y) printResult("reduceByKey", rdd12) // 結(jié)果：reduceByKey >> List((0,18), (2,15), (1,12))

• sortByKey

sortByKey(ascending, numPartitions)是對RDD中的數(shù)據(jù)集進(jìn)行排序操作，對(K,V)類型的數(shù)據(jù)按照K進(jìn)行排序，其中K需要實現(xiàn)Ordered方法。

第一個參數(shù)是ascending，該參數(shù)決定排序后RDD中的元素是升序還是降序，默認(rèn)是true,按升序排序。
第二個參數(shù)是numPartitions，即排序分區(qū)的并行任務(wù)個數(shù)。

val rdd15 = rddMap.sortByKey(false) printResult("sortByKey", rdd15) // 結(jié)果：sortByKey >> List((2,2), (2,5), (2,8), (1,1), (1,4), (1,7), (0,3), (0,6), (0,9))

• aggregateByKey

aggregateByKey(zeroValue, numPartitions)(seqOp: (U, V) => U,combOp: (U, U) => U)和reduceByKey的不同在于:reduceByKey輸入/輸出都是(K,V)，而aggregateByKey輸出是(K,U)，可以不同于輸入(K,U)。
ggregateByKey的三個參數(shù)如下： ? ? ? ?

zeroValue:U，初始值，比如初始值為0或空列表。 ? ? ? ?

numPartitions:指定并行任務(wù)的分區(qū)數(shù)。 ? ? ? ?

seqOp:(U,V)=>U,seq操作符，描述如何將T合并入U,比如如何將item合并到列表中。 ? ? ? ?

combOp: (U, U) => U,comb操作符，描述如何合并兩個U,比如合并兩個列表。 ? ? ? ?

所以，可以將aggregateByKey函數(shù)抽象成更高級的，更靈活的reduce和group的組合。

val rdd13 = rddMap.aggregateByKey(0)(seqOp = (x, y) => {math.max(x, y)},combOp = (x, y) => {x + y}) printResult("aggregateByKey", rdd13) // 結(jié)果：aggregateByKey >> List((0,12), (2,10), (1,11))

• combineByKey

combineByKey(createCombiner: V => C,mergeValue: (C, V) => C,mergeCombiners: (C, C) => C,numPartitions: Int)是對RDD中的數(shù)據(jù)按照Key進(jìn)行聚合操作。聚合操作的邏輯是通過自定義函數(shù)提供給combineByKey。?把(K,V)類型的RDD轉(zhuǎn)換為(K,C)類型的RDD,C和Vk可以不一樣。
combineyey的三個參數(shù)如下： ?

?createCombiner:在遍歷(k,v)時，如果combineByKey的第一次遇到值為K的key(類型K),那么將對這個(K,V)調(diào)用createCombiner函數(shù)，將V轉(zhuǎn)換為c(類型C,聚合對象的類型)。 ? ? ?

mergeValue：在遍歷(k,v)時，如果comineByKey不是第一次遇到值為k的key(類型為K),那么將對這個(k,v)調(diào)用mergeValue函數(shù)，它的作用是將v累加到聚合對象(類型C)中，mergeValue的類型是(C，V)=>C，參數(shù)中的C遍歷到此處的聚合對象，然后對v進(jìn)行聚合得到新的聚合對象值。 ? ? ?

mergeCombiners：combineByKey實在分布式環(huán)境中執(zhí)行的，RDD的每個分區(qū)單獨進(jìn)行combineBykey操作，最后需要對各個分區(qū)的結(jié)果進(jìn)行最后的聚合。它的函數(shù)類型是(C,C)=>C,每個參數(shù)是分區(qū)聚合得到的聚合對象。?

val rdd14 = rdd1.map(item => (item % 4, item)).mapValues(v => v.toDouble).combineByKey((v: Double) => (v, 1),(c: (Double, Int), v: Double) => (c._1 + v, c._2 + 1),(c1: (Double, Int), c2: (Double, Int)) => (c1._1 + c2._1, c1._2 + c2._2)) printResult("combineByKey", rdd14) // 結(jié)果：combineByKey >> List((0,(12.0,2)), (2,(8.0,2)), (1,(15.0,3)), (3,(10.0,2)))

• jion

join(other, partitioner)是連接操作，將數(shù)據(jù)的數(shù)據(jù)集(K,V)和另外一個數(shù)據(jù)集(K,W)進(jìn)行join,得到(K,(V,W))。該操作是對于相同K的V和W集合進(jìn)行笛卡爾積操作，也即V和W的所有組合。連接操作除了join外，還有左連接，右連接，全連接操作函數(shù)：leftOuterJoin，rightOuterJoin和fullOuterJoin，它們的用法基本上是一樣的。

val rddMap1 = rdd1.map(item => (item % 4, item)) val rdd16 = rddMap.join(rddMap1) printResult("join", rdd16) // 結(jié)果：join >> List((0,(3,4)), (0,(3,8)), (0,(6,4)), (0,(6,8)), (0,(9,4)), (0,(9,8)), (2,(2,2)), (2,(2,6)), (2,(5,2)), (2,(5,6)), (2,(8,2)), (2,(8,6)), (1,(1,1)), (1,(1,5)), (1,(1,9)), (1,(4,1)), (1,(4,5)), (1,(4,9)), (1,(7,1)), (1,(7,5)), (1,(7,9)))

• pipe

pipe(command: Seq[String],env: Map[String, String])是以shell命令處理RDD數(shù)據(jù)。

val rdd19: RDD[String] = rdd1.pipe("head -n 1") printResult("pipe", rdd19) // 結(jié)果：pipe >> List(1,4,7)

• subtract

subtract(other: RDD[T], numPartitions: Int)是RDD對other數(shù)據(jù)集進(jìn)行減法操作，將輸入的rdd中的元素減去other中的元素，得到他們差值的一個新的RDD。

val rdd21 = rdd1.subtract(sc.parallelize(1 to 5)) printResult("subtract", rdd21) // 結(jié)果：subtract >> List(6, 9, 7, 8)

• zip

zip[U: ClassTag](other: RDD[U]): RDD[(T, U)]是對兩個RDD進(jìn)行拉鏈操作，得到一個新的RDD[T,U]。拉鏈操作還包括：zipWitIndex和zipPartitinos。

val rdd22 = rdd1.repartition(3).zip(sc.parallelize(Array("A", "B", "C", "D", "E", "F", "G", "H", "I"), 3)) printResult("zip", rdd22) // 結(jié)果: zip >> List((3,A), (6,B), (8,C), (1,D), (4,E), (9,F), (2,G), (5,H), (7,I))

• coalesce 和?repartition

coalesce(numPartitions: Int)是對RDD進(jìn)行重分區(qū)，默認(rèn)不進(jìn)行shuffle，且該RDD的分區(qū)個數(shù)等于numPartitions個數(shù)。

repartition(numPartitions: Int)是將RDD進(jìn)行重新分區(qū)，分區(qū)過程中會進(jìn)行shuffle，調(diào)整分區(qū)數(shù)量為numPartitions。

總結(jié)

以上是生活随笔為你收集整理的[scala-spark]10. RDD转换操作的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。