RDD基本轉換：

RDD[U] map(f: T => U)

T：原RDD中元素型別

U：新RDD中元素型別

函式將T元素轉換為新的U元素

rdd.map(x => x + 1)

{1, 2, 3, 3}

=>{2, 3, 4, 4}

RDD[U] flatMap(f: T => TraversableOnce[U])

TraversableOnce：集合與迭代器的父類

函式將T元素轉換為含有新型別U元素的集合，並將這些集合展平（兩層轉換成一層）後的元素形成新的RDD

rdd.flatMap(x => x.to(3))

{1, 2, 3, 3}

=>{1, 2, 3, 2, 3, 3, 3}

RDD[T] filter(f: T => Boolean)

函式對每個元素進行過濾，通過的元素形成新的RDD

rdd.filter(x => x != 1)

{1, 2, 3, 3}

=>{2, 3, 3}

RDD[T] distinct()

去重

rdd.distinct()

{1, 2, 3, 3}

=>{1, 2, 3}

RDD[U] mapPartitions(f: Iterator[T] => Iterator[U])

與map一樣，只是轉換時是以分割槽為單位，將一個分割槽所有元素包裝成Iterator一次性傳入函式進行處理，而不像map函式那樣每個元素都會呼叫一個函式，即這裡有幾個分割槽則才呼叫幾次函式

假設有N個元素，有M個分割槽，那麼map的函式的將被呼叫N次,而mapPartitions被呼叫M次

valarr = Array(1, 2, 3, 4, 5)

valrdd = sc.parallelize(arr, 2)

rdd.mapPartitions((it: Iterator[Int]) => { var l = List[Int](); it.foreach((e: Int) => l = e * 2 :: l); l.iterator })

=>{2, 4, 6, 8, 10}

RDD[U] mapPartitionsWithIndex(f: (Int, Iterator[T]) => Iterator[U])

與mapPartitions類似，不同的時函式多了個分割槽索引的引數

RDD[T] union(other: RDD[T])

兩個RDD 並集，包括重複的元素

rdd.union(otherRdd)

  { 1, 2, 2, 3, 3}

  { 3, 4, 5}

=>{1, 2, 2, 3, 3, 3, 4, 5}

RDD[T] intersection(other: RDD[T])

兩個RDD 交集

rdd.intersection(otherRdd)

  { 1, 2, 2, 3, 3}

  { 3, 4, 5}

=>{3}

RDD[T] subtract(other: RDD[T])

兩個RDD相減

rdd.subtract(otherRdd)

  { 1, 2, 2, 3, 3}

  { 3, 4, 5}

=>{1, 2, 2}

RDD[(T, U)] cartesian(other: RDD[U])

兩個RDD相減笛卡兒積

rdd.cartesian(otherRdd)

  { 1, 2 }

  { 3, 4}

=>{(1,3),(1,4),(2,3),(2,4)}

RDD[T] sortBy( f: (T) => K, ascending: Boolean,numPartitions: Int)

根據轉換後的值進行排序，傳入的是一個(T) => K 轉換函式

rdd.sortBy(_._2, false, 1)

這裡根據value進行降序排序

  {("leo", 65), ("tom", 50), ("marry", 100), ("jack", 80)}

=>{("marry", 100),("jack", 80),("leo", 65), ("leo", 65)}

RDD[Array[T]] glom()

將RDD的每個分割槽中的型別為T的元素轉換換陣列Array[T]

valarr = Array(1, 2, 3, 4, 5)

valrdd = sc.parallelize(arr, 2)

valarrRDD = rdd.glom()arrRDD.foreach { (arr: Array[Int]) => { println("[ " + arr.mkString(" ") + " ]"); } }

=>[ 1 2 ],[ 3 4 5 ]

鍵-值RDD轉換：

RDD[(K, U)] mapValues[U](f: V => U)

K：key型別

V：value型別

將value轉換為新的U元素，Key不變

rdd.mapValues(_ + 1)

{"class1", 80), ("class2", 70)}

=>{"class1", 81), ("class2", 71)}

RDD[(K, U)] flatMapValues(f: V => TraversableOnce[U])

對[K,V]型資料中的V值flatmap操作

rdd.flatMapValues(_.toCharArray())

{ (1, "ab"), (2, "bc")}

=>{(1, 'a'), (1, 'b'), (2, 'b'), (2, 'c')}

RDD[(K, Iterable[V])] groupByKey()

根據key進行分組，同一組的元素組成Iterable<V>，並以(key, Iterable<V>)元組型別為元素作為新的RDD返回

rdd.groupByKey()

{("class1", 80), ("class2", 75), ("class1", 90), ("class2", 60)}

=>{("class1",[80,90]),("class2",[75,60])}

RDD[(K, Iterable[T])] groupBy(f: T => K)

T：原RDD元素型別

K：新RDD中元素Key的型別

根據函式將元素T對映成相應K後，以此K進行分組

rdd.groupBy({ case 1 => 1; case 2 => 2; case "二" => 2 })

  { 1, 2, "二" }

=>{(1,[1]),(2,[2, "二"])}

RDD[(K, V)] reduceByKey(func: (V, V) => V)

先根據key進行分組，再對同一組中的的value進行reduce操作：第一次呼叫函式時傳入的是兩個Key所對應的value，從第二次往後，傳入的兩個引數中的第一個為上次函式計算的結果，第二個引數為其它Key的value

rdd. reduceByKey(_ + _)

{("class1", 80), ("class2", 75), ("class1", 90), ("class2", 60)}

=>{("class1", 170),("class2", 135)}

RDD[(K, V)] sortByKey()

根據key的大小進行排序（注：並不是先以Key進行分組，再對組類進行排序，而是直接根據Key的值進行排序）

rdd.sortByKey(false)

{(65, "leo"), (50, "tom"),(100, "marry"), (85, "jack")}

=>{(100, "marry"),(85, "jack"),(65, "eo"),(50, "tom")}

RDD[(K, V)] foldByKey(zeroValue: V)(func: (V, V) => V):

zeroValue：每個分割槽相同Key累計時的初始值，以及不同分割槽相同Key合併時的初始值

e.g., Nil for list concatenation, 0 for addition, or 1 for multiplication

對每個value先進行func操作，且funcfoldByKey函式是通過呼叫函式實現的。

zeroVale：對V進行初始化，實際上是通過CombineByKey的createCombiner實現的  V =>  (zeroValue,V)，再通過func函式對映成新的值，即func(zeroValue,V)

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

valpeople = List(("Mobin", 1), ("Lucy", 2), ("Amy", 3), ("Amy", 4), ("Lucy", 5))

valrdd = sc.parallelize(people,2)

valfoldByKeyRDD = rdd.foldByKey(10)((v1, v2) => { println(v1 + " + " + v2 + " = " + (v1 + v2)); v1 + v2 }) //先對每個V都加10，再對相同Key的value值相加

foldByKeyRDD.foreach(println)

//處理第一個分割槽資料

10 + 1 = 11 // ("Mobin", 1)

10 + 2 = 12 // ("Lucy", 2)

=====================

//處理第二個分割槽資料

10 + 3 = 13 // ("Amy", 3)

13 + 4 = 17 // ("Amy", 4)同分區同Key的Val先合併

10 + 5 = 15 // ("Lucy", 5)

=====================

//將不同分割槽相同Key的Value合併起來

12 + 15 = 27 // "Lucy"跨分割槽，所以需合併

(Amy,17)

(Mobin,11)

(Lucy,27)

RDD[(K, (V, Option[W]))] leftOuterJoin[W](other: RDD[(K, W)]):

左外連線，包含左RDD的所有資料，如果右邊沒有與之匹配的用None表示

valarr = List(("A", 1), ("A", 2), ("B", 1))

valarr1 = List(("A", "A1"), ("A", "A2"))

valrdd = sc.parallelize(arr, 2)

valrdd1=sc.parallelize(arr1, 2)

valleftOutJoinRDD = rdd.leftOuterJoin(rdd1)

leftOutJoinRDD.foreach(println)

=>

(B,(1,None))

(A,(1,Some(A1)))

(A,(1,Some(A2)))

(A,(2,Some(A1)))

(A,(2,Some(A2)))

RDD[(K, (Option[V], W))] rightOuterJoin[W](other: RDD[(K, W)])

右外連線，包含右RDD的所有資料，如果左邊沒有與之匹配的用None表示

valarr = List(("A", 1), ("A", 2))

valarr1 = List(("A", "A1"), ("A", "A2"), ("B", 1))

valrdd = sc.parallelize(arr, 2)

valrdd1 = sc.parallelize(arr1, 2)

valleftOutJoinRDD = rdd.rightOuterJoin

相關推薦