RDD：彈性分散式資料集，是一種特殊集合 ‚ 支援多種來源 ‚ 有容錯機制 ‚ 可以被快取 ‚ 支援並行操作，一個RDD代表一個分割槽裡的資料集
RDD有兩種操作運算元：

        Transformation（轉換）：Transformation屬於延遲計算，當一個RDD轉換成另一個RDD時並沒有立即進行轉換，僅僅是記住       了資料集的邏輯操作
         Ation（執行）：觸發Spark作業的執行，真正觸發轉換運算元的計算

 
本系列主要講解Spark中常用的函式操作：
         1.RDD基本轉換
         2.鍵-值RDD轉換
         3.Action操作篇

 

本節所講函式

1.mapValus

2.flatMapValues

3.comineByKey

4.foldByKey

5.reduceByKey

6.groupByKey

7.sortByKey

8.cogroup

9.join

10.LeftOutJoin

11.RightOutJoin

 

1.mapValus(fun):對[K,V]型資料中的V值map操作

(例1)：對每個的的年齡加2

1 2 3 4 5 6 7 8 9 10

object MapValues {   def main(args: Array[String]) {     val conf = new SparkConf().setMaster("local").setAppName( "map")     val sc = new SparkContext(conf)     val list = List(("mobin",22),("kpop",20),("lufei",23))     val rdd = sc.parallelize(list)     val mapValuesRDD = rdd.mapValues(_+2)     mapValuesRDD.foreach(println)   } }

輸出：

(mobin,24)
(kpop,22)
(lufei,25)

(RDD依賴圖：紅色塊表示一個RDD區，黑色塊表示該分割槽集合，下同)

 

 

2.flatMapValues(fun)：對[K,V]型資料中的V值flatmap操作

(例2):

1 2 3 4

//省略<br>val list = List(("mobin",22),("kpop",20),("lufei",23)) val rdd = sc.parallelize(list) val mapValuesRDD = rdd.flatMapValues(x => Seq(x,"male")) mapValuesRDD.foreach(println)

輸出：

(mobin,22)
(mobin,male)
(kpop,20)
(kpop,male)
(lufei,23)
(lufei,male)

如果是mapValues會輸出：

(mobin,List(22, male))
(kpop,List(20, male))
(lufei,List(23, male))

（RDD依賴圖）

 

 

3.comineByKey(createCombiner,mergeValue,mergeCombiners,partitioner,mapSideCombine)

 

   comineByKey(createCombiner,mergeValue,mergeCombiners,numPartitions)

 

   comineByKey(createCombiner,mergeValue,mergeCombiners)

 

createCombiner:在第一次遇到Key時建立組合器函式，將RDD資料集中的V型別值轉換C型別值（V => C），

如例3:

mergeValue：合併值函式，再次遇到相同的Key時，將createCombiner道理的C型別值與這次傳入的V型別值合併成一個C型別值（C,V）=>C，

如例3：

mergeCombiners:合併組合器函式，將C型別值兩兩合併成一個C型別值

如例3：

 

partitioner：使用已有的或自定義的分割槽函式，預設是HashPartitioner

 

mapSideCombine：是否在map端進行Combine操作,預設為true

 

注意前三個函式的引數型別要對應；第一次遇到Key時呼叫createCombiner，再次遇到相同的Key時呼叫mergeValue合併值

 

（例3）：統計男性和女生的個數，並以（性別，（名字，名字....），個數）的形式輸出

1 2 3 4 5 6 7 8 9 10 11 12 13 14

object CombineByKey {   def main(args: Array[String]) {     val conf = new SparkConf().setMaster("local").setAppName("combinByKey")     val sc = new SparkContext(conf)     val people = List(("male", "Mobin"), ("male", "Kpop"), ("female", "Lucy"), ("male", "Lufei"), ("female", "Amy"))     val rdd = sc.parallelize(people)     val combinByKeyRDD = rdd.combineByKey(       (x: String) => (List(x), 1),       (peo: (List[String], Int), x : String) => (x :: peo._1, peo._2 + 1),       (sex1: (List[String], Int), sex2: (List[String], Int)) => (sex1._1 ::: sex2._1, sex1._2 + sex2._2))     combinByKeyRDD.foreach(println)     sc.stop()   } }

輸出：

(male,(List(Lufei, Kpop, Mobin),3))
(female,(List(Amy, Lucy),2))

過程分解：

Partition1:
K="male"  -->  ("male","Mobin")  --> createCombiner("Mobin") =>  peo1 = (  List("Mobin") , 1 )
K="male"  -->  ("male","Kpop")  --> mergeValue(peo1,"Kpop") =>  peo2 = (  "Kpop"  ::  peo1_1 , 1 + 1 )    //Key相同呼叫mergeValue函式對值進行合併
K="female"  -->  ("female","Lucy")  --> createCombiner("Lucy") =>  peo3 = (  List("Lucy") , 1 )
 
Partition2:
K="male"  -->  ("male","Lufei")  --> createCombiner("Lufei") =>  peo4 = (  List("Lufei") , 1 )
K="female"  -->  ("female","Amy")  --> createCombiner("Amy") =>  peo5 = (  List("Amy") , 1 )
 
Merger Partition:
K="male" --> mergeCombiners(peo2,peo4) => (List(Lufei,Kpop,Mobin))
K="female" --> mergeCombiners(peo3,peo5) => (List(Amy,Lucy))

（RDD依賴圖）

 

4.foldByKey(zeroValue)(func)

 

  foldByKey(zeroValue,partitioner)(func)

 

  foldByKey(zeroValue,numPartitiones)(func)

 

foldByKey函式是通過呼叫CombineByKey函式實現的

 

zeroVale：對V進行初始化，實際上是通過CombineByKey的createCombiner實現的  V =>  (zeroValue,V)，再通過func函式對映成新的值，即func(zeroValue,V),如例4可看作對每個V先進行  V=> 2 + V  

 

func: Value將通過func函式按Key值進行合併（實際上是通過CombineByKey的mergeValue，mergeCombiners函式實現的，只不過在這裡，這兩個函式是相同的）

例4：

1 2 3 4 5

//省略     val people = List(("Mobin", 2), ("Mobin", 1), ("Lucy", 2), ("Amy", 1), ("Lucy", 3))     val rdd = sc.parallelize(people)     val foldByKeyRDD = rdd.foldByKey(2)(_+_)     foldByKeyRDD.foreach(println)

輸出：

(Amy,2)
(Mobin,4)
(Lucy,6)

先對每個V都加2，再對相同Key的value值相加。

 

 

5.reduceByKey(func,numPartitions):按Key進行分組，使用給定的func函式聚合value值, numPartitions設定分割槽數，提高作業並行度

例5

1 2 3 4 5 6

//省略 val arr = List(("A",3),("A",2),("B",1),("B",3)) val rdd = sc.parallelize(arr) val reduceByKeyRDD = rdd.reduceByKey(_ +_) reduceByKeyRDD.foreach(println) sc.stop

輸出：

(A,5)
(A,4)

（RDD依賴圖）

 

6.groupByKey(numPartitions):按Key進行分組，返回[K,Iterable[V]]，numPartitions設定分割槽數，提高作業並行度

例6：

1 2 3 4 5 6

//省略 val arr = List(("A",1),("B",2),("A",2),("B",3)) val rdd = sc.parallelize(arr) val groupByKeyRDD = rdd.groupByKey() groupByKeyRDD.foreach(println) sc.stop

輸出：

(B,CompactBuffer(2, 3))
(A,CompactBuffer(1, 2))

 

以上foldByKey,reduceByKey,groupByKey函式最終都是通過呼叫combineByKey函式實現的

 

7.sortByKey(accending，numPartitions):返回以Key排序的（K,V）鍵值對組成的RDD，accending為true時表示升序，為false時表示降序，numPartitions設定分割槽數，提高作業並行度

例7：

1 2 3 4 5 6

//省略sc val arr = List(("A",1),("B",2),("A",2),("B",3)) val rdd = sc.parallelize(arr) val sortByKeyRDD = rdd.sortByKey() sortByKeyRDD.foreach(println) sc.stop

輸出：

(A,1)
(A,2)
(B,2)
(B,3)

 

8.cogroup(otherDataSet，numPartitions)：對兩個RDD(如:(K,V)和(K,W))相同Key的元素先分別做聚合，最後返回(K,Iterator<V>,Iterator<W>)形式的RDD,numPartitions設定分割槽數，提高作業並行度

例8：

1 2 3 4 5 6 7 8

//省略 val arr = List(("A", 1), ("B", 2), ("A", 2), ("B", 3)) val arr1 = List(("A", "A1"), ("B", "B1"), ("A", "A2"), ("B", "B2")) val rdd1 = sc.parallelize(arr, 3) val rdd2 = sc.parallelize(arr1, 3) val groupByKeyRDD = rdd1.cogroup(rdd2) groupByKeyRDD.foreach(println) sc.stop

輸出：

(B,(CompactBuffer(2, 3),CompactBuffer(B1, B2)))
(A,(CompactBuffer(1, 2),CompactBuffer(A1, A2)))

（RDD依賴圖）

 

9.join(otherDataSet,numPartitions):對兩個RDD先進行cogroup操作形成新的RDD，再對每個Key下的元素進行笛卡爾積，numPartitions設定分割槽數，提高作業並行度

例9

1 2 3 4 5 6 7

//省略 val arr = List(("A", 1), ("B", 2), ("A", 2), ("B", 3)) val arr1 = List(("A", "A1"), ("B", "B1"), ("A", "A2"), ("B", "B2")) val rdd = sc.parallelize(arr, 3) val rdd1 = sc.parallelize(arr1, 3) val groupByKeyRDD = rdd.join(rdd1) groupByKeyRDD.foreach(println)

輸出：

(B,(2,B1))
(B,(2,B2))
(B,(3,B1))
(B,(3,B2))
 
(A,(1,A1))
(A,(1,A2))
(A,(2,A1))
(A,(2,A2)

（RDD依賴圖）

 

10.LeftOutJoin(otherDataSet，numPartitions):左外連線，包含左RDD的所有資料，如果右邊沒有與之匹配的用None表示,numPartitions設定分割槽數，提高作業並行度

例10：

1 2 3 4 5 6 7 8

//省略 val arr = List(("A", 1), ("B", 2), ("A", 2), ("B", 3),("C",1)) val arr1 = List(("A", "A1"), ("B", "B1"), ("A", "A2"), ("B", "B2")) val rdd = sc.parallelize(arr, 3) val rdd1 = sc.parallelize(arr1, 3) val leftOutJoinRDD = rdd.leftOuterJoin(rdd1) leftOutJoinRDD .foreach(println) sc.stop

輸出：

(B,(2,Some(B1)))
(B,(2,Some(B2)))
(B,(3,Some(B1)))
(B,(3,Some(B2)))
 
(C,(1,None))
 
(A,(1,Some(A1)))
(A,(1,Some(A2)))
(A,(2,Some(A1)))
(A,(2,Some(A2)))

 

11.RightOutJoin(otherDataSet, numPartitions):右外連線，包含右RDD的所有資料，如果左邊沒有與之匹配的用None表示,numPartitions設定分割槽數，提高作業並行度

例11：

1 2 3 4 5 6 7 8

//省略 val arr = List(("A", 1), ("B", 2), ("A", 2), ("B", 3)) val arr1 = List(("A", "A1"), ("B", "B1"), ("A", "A2"), ("B", "B2"),("C","C1")) val rdd = sc.parallelize(arr, 3) val rdd1 = sc.parallelize(arr1, 3) val rightOutJoinRDD = rdd.rightOuterJoin(rdd1) rightOutJoinRDD.foreach(println) sc.stop

輸出：

(B,(Some(2),B1))
(B,(Some(2),B2))
(B,(Some(3),B1))
(B,(Some(3),B2))
 
(C,(None,C1))
 
(A,(Some(1),A1))
(A,(Some(1),A2))
(A,(Some(2),A1))
(A,(Some(2),A2))

 

以上例子原始碼地址：https://github.com/Mobin-F/SparkExample/tree/master/src/main/scala/com/mobin/SparkRDDFun/TransFormation/RDDBase