常用SparkRDD容易混淆的運算元區別

1.map與flatMap的區別

# 初始化資料
val rdd1 = sc.parallelize(Array("hello world","i love you"))

• map

# map運算元
rdd1.map(_.split(" ")).collect

# map運算元結果輸出
res0: Array[Array[String]] = Array(Array(hello, world), Array(i, love, you))

• flatMap

# flatMap運算元
rdd1.flatMap(_.split(" "
 
)).collect

# flatMap運算元結果輸出
res1: Array[String] = Array(hello, world, i, love, you)

flatMap是將資料先進行map轉化，在通過flattern對map結果進行’壓平’。也就是將map轉化的2個Array壓平處理。

flatMap的效率要優於map

2.map與mapPartitions的區別

map是對rdd中的每一個元素進行操作。

mapPartitions(foreachPartition)則是對rdd中的每個分割槽的迭代器進行操作。

如果在map過程中需要頻繁建立額外的物件(例如將rdd中的資料通過jdbc寫入資料庫，map需要為每個元素建立一個連結而mapPartition為每個partition建立一個連結),則mapPartitions效率比map高的多

。

• map示例

# 初始化資料，設定3個分割槽
val rdd = sc.parallelize(Array(1,2,3,4,5,6,7,8,9),3)


rdd.map(x=>{println(x);x*2}).collect

輸出結果：

• mapPartitions示例

rdd.mapPartitions(x=>{println("-----------");for(i<-x)yield i*2}).collect

輸出結果：

通過對比，我們可以發現，mapPartitions在每個分割槽只會呼叫一次。而map每次都會被呼叫。所以在特定的場景mapPartitions效率比map高的多

。

3.mapPartitionsWithIndex示例

• mapPartitionsWithIndex

類似於mapPartitions，但取兩個引數。

第一個引數是分割槽的索引（index），第二個引數是通過這個分割槽中的所有項的迭代器(iterator)。

輸出是一個迭代器，它包含在應用函式編碼的任何轉換之後的專案列表。

# 初始化資料
val rdd = sc.makeRDD(1 to 10, 3)

# mapPartitionsWithIndex方法是呼叫，傳入一個函式，函式引數1為index，引數2為iterator
rdd.mapPartitionsWithIndex((index, iter) => iter.map((index, _))).foreach(println)

輸出結果：

4.mapValues與flatMapValues

• mapValues

# 初始化rdd
val rdd = sc.makeRDD(List("hello spark", "hello java", "hello python", "hello r","hello scala"), 2)


rdd.map(x => (x.length % 3, x)).mapValues(x => x.split(" ")).collect()

輸出結果：

• flatMapValues

# 初始化rdd
val rdd = sc.makeRDD(List("hello spark", "hello java", "hello python", "hello r","hello scala"), 2)

rdd.map(x => (x.length % 3, x)).flatMapValues(x => x.split(" ")).collect()

輸出結果：

flatMapValues是把map中的value進行過函式操作後，再將資料結構壓平。

5.coalesce、repartition與partitionBy

• coalesece

該函式用於將RDD進行重分割槽，使用HashPartitioner進行分割槽。

接受2個引數，第一個引數為重分割槽的數目，第二個為是否進行shuffle，預設為false;

val rdd1 = sc.parallelize(1 to 12,3)
println(rdd1.partitions.length) //輸出結果 3

val rdd2 = rdd1.coalesce(2)
println(rdd2.partitions.length) // 輸出結果 2

//如果重分割槽的數目大於原來的分割槽數，那麼必須指定shuffle引數為true，//否則，分割槽數不變
val rdd3 = rdd1.coalesce(4)
println(rdd3.partitions.length) // 輸出結果 3

val rdd4 = rdd1.coalesce(4,true)
println(rdd4.partitions.length) // 輸出結果 4

• repartition

該函式其實就是coalesce函式第二個引數為true的實現

def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
    coalesce(numPartitions, shuffle = true)
  }

示例：

val rdd1 = sc.parallelize(1 to 12,3)

val rdd5 = rdd1.repartition(4)

println(rdd5.partitions.length) // 輸出結果 4

• partitionBy

partitionBy：表示重新分割槽，不是僅僅是設定分割槽數

var rdd1 = sc.makeRDD(Array((1, "A"), (2, "B"), (3, "C"), (4, "D")), 2)
rdd1.mapPartitionsWithIndex { (index, iter) => for (i <- iter) yield (index, i) }.foreach(println)

# 重新設定分割槽
var rdd2 = rdd1.partitionBy(new org.apache.spark.HashPartitioner(2))
rdd2.mapPartitionsWithIndex { (index, iter) => for (i <- iter) yield (index, i) }.foreach(println)

結果如圖：4.13圖

6.union、distinct、intersection、subtract

• union

求並集

val rdd1 = sc.parallelize(List(5, 6, 4, 3))
val rdd2 = sc.parallelize(List(1, 2, 3, 4))
rdd1.union(rdd2).collect

# 輸出結果，這個結果不會去重
res52: Array[Int] = Array(5, 6, 4, 3, 1, 2, 3, 4)

• distinct

去重

val rdd1 = sc.parallelize(List(5, 6, 4, 3))
val rdd2 = sc.parallelize(List(1, 2, 3, 4))
val rdd3 = rdd1.union(rdd2)

rdd3.distinct.collect

# 輸出結果，結果去重
res53: Array[Int] = Array(4, 6, 2, 1, 3, 5)

• intersection

求交集

val rdd1 = sc.parallelize(List(5, 6, 4, 3))
val rdd2 = sc.parallelize(List(1, 2, 3, 4))

rdd1.intersection(rdd2).collect

# 輸出結果：
res54: Array[Int] = Array(4, 3)

• subtract

求差集

val rdd1 = sc.parallelize(List(5, 6, 4, 3))
val rdd2 = sc.parallelize(List(1, 2, 3, 4))

rdd1.subtract(rdd2).collect

# 輸出結果
res55: Array[Int] = Array(5, 6)



rdd2.subtract(rdd1).collect

# 輸出結果
res56: Array[Int] = Array(1, 2)

• subtractByKey

subtractByKey和基本轉換操作中的subtract類似,只不過這裡是針對K的

val rdd1 = sc.makeRDD(Array(("A", "1"), ("B", "2"), ("B", "3"), ("C", "3")), 2)
val rdd2 = sc.makeRDD(Array(("A", "a"), ("C", "c"), ("D", "d")), 2)

rdd1.subtractByKey(rdd2).collect

# 輸出結果：因為rdd2的key中有A,C，所以差集就是B
res57: Array[(String, String)] = Array((B,2), (B,3))


rdd2.subtractByKey(rdd1).collect

# 輸出結果：
res58: Array[(String, String)] = Array((D,d))

7.groupByKey、reduceByKey、foldByKey、aggregateByKey

• groupByKey

val rdd1 = sc.parallelize(List(("tom", 1), ("jerry", 3), ("kitty", 2), ("shuke", 1)))
val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 3), ("shuke", 2), ("kitty", 5)))
//求並集
val rdd3 = rdd1 union rdd2

rdd3.groupByKey().collect

# 輸出結果：
res89: Array[(String, Iterable[Int])] = Array((tom,CompactBuffer(1, 3)), (jerry,CompactBuffer(3, 2)), (shuke,CompactBuffer(1, 2)), (kitty,CompactBuffer(2, 5)))


rdd3.groupByKey().map(t=>(t._1,t._2.sum)).collect

# 輸出結果：
res90: Array[(String, Int)] = Array((tom,4), (jerry,5), (shuke,3), (kitty,7))

• reduceByKey

val rdd1 = sc.parallelize(List(("tom", 1), ("jerry", 3), ("kitty", 2), ("shuke", 1)))
val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 3), ("shuke", 2), ("kitty", 5)))

val rdd3 = rdd1.union(rdd2)
//按key進行聚合
rdd3.reduceByKey(_ + _).collect

# 輸出結果：
res91: Array[(String, Int)] = Array((tom,4), (jerry,5), (shuke,3), (kitty,7))

groupByKey和reduceByKey都可以通過key來進行聚合。不同的是2者，聚合的過程是不一樣的。

reduceByKey現在map過程中先進行聚合，再到reduce端聚合，減少資料太大帶來的壓力，減小RPC過程中的傳輸壓力。

groupByKey是直接在reduce端進行聚合的，所以效率比reduceByKey低。

推薦使用reduceByKey，因為效率比groupByKey高

• foldByKey

該函式用於RDD[K,V]根據K將V做摺疊、合併處理，其中的引數zeroValue表示先根據對映函式將zeroValue應用於V,進行初始化V,再將對映函式應用於初始化後的V.

val rdd1 = sc.parallelize(List(("tom", 1), ("jerry", 3), ("kitty", 2), ("shuke", 1)))
val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 3), ("shuke", 2), ("kitty", 5)))

val rdd3 = rdd1.union(rdd2)

rdd3.foldByKey(0)(_ + _).collect

# 輸出結果：
res84: Array[(String, Int)] = Array((tom,4), (jerry,5), (shuke,3), (kitty,7))

foldByKey和reduceByKey的功能是相似的，都是在map端先進行聚合，再到reduce聚合。不同的是flodByKey需要傳入一個引數。該引數是計算的初始值。

• aggregateByKey

aggregateByKey函式對PairRDD中相同Key的值進行聚合操作，在聚合過程中同樣使用了一箇中立的初始值。

def aggregateByKey[U](zeroValue: U)(seqOp: (U, V) => U, combOp: (U, U) => U)(implicit arg0: ClassTag[U]): RDD[(K, U)]
def aggregateByKey[U](zeroValue: U, numPartitions: Int)(seqOp: (U, V) => U, combOp: (U, U) => U)(implicit arg0: ClassTag[U]): RDD[(K, U)]
def aggregateByKey[U](zeroValue: U, partitioner: Partitioner)(seqOp: (U, V) => U, combOp: (U, U) => U)(implicit arg0: ClassTag[U]): RDD[(K, U)]

示例：

val rdd1 = sc.parallelize(List(("tom", 1), ("jerry", 3), ("kitty", 2), ("shuke", 1)))
val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 3), ("shuke", 2), ("kitty", 5)))

val rdd3 = rdd1.union(rdd2)

# aggregateByKey(0)(math.max(_,_),_+_)
# (0)：0是傳入的初始值
# math.max(_,_)：是一個函式，表示傳入的資料與初始值比較，取最小值
# _+_：和reduceByKey中一樣，表示`(x,y)=>x+y`這樣的函式

rdd3.aggregateByKey(0)(math.max(_,_),_+_).collect

# 輸出結果：
res93: Array[(String, Int)] = Array((tom,4), (jerry,5), (shuke,3), (kitty,7))

aggregateByKey會先在本節點內先聚合，然後再聚合所有節點的結果。

8.zip函式與zipPartitions函式

• zip函式

zip函式用於將兩個RDD組合成Key/Value形式的RDD

var rdd1 = sc.makeRDD(1 to 5, 2)
var rdd2 = sc.makeRDD(Seq("A", "B", "C", "D", "E"), 2)

rdd1.zip(rdd2).foreach(println)

# 輸出結果：
(1,A)
(2,B)
(3,C)
(4,D)
(5,E)


rdd2.zip(rdd1).foreach(println)

# 輸出結果：
(A,1)
(B,2)
(C,3)
(D,4)
(E,5)

這裡需要注意的是：

1.如果兩個RDD分割槽數不同，則丟擲異常:Can’t zip RDDs with unequal numbers of partitions

2.如果兩個RDD的元素個數不同，則丟擲異常：Can only zip RDDs with same number of elements in each partition

• zipPartitions函式

zipPartitions函式將多個RDD按照partition組合成為新的RDD。

該函式需要組合的RDD具有相同的分割槽數，但對於每個分割槽內的元素數量沒有要求。

2個RDD進行zipPartitions操作

var rdd1 = sc.makeRDD(1 to 5, 2)
var rdd2 = sc.makeRDD(Seq("A", "B", "C", "D", "E"), 2)

rdd1.zipPartitions(rdd2) {
  (rdd1Iter, rdd2Iter) => {
    var result = List[String]()
    while (rdd1Iter.hasNext && rdd2Iter.hasNext) {
      result ::= (rdd1Iter.next() + "_" + rdd2Iter.next())
    }
    result.iterator
  }
}.foreach(println)


# 輸出結果：
2_B
1_A
5_E
4_D
3_C

3個RDD進行zipPartitions操作

var rdd1 = sc.makeRDD(1 to 5, 2)
var rdd2 = sc.makeRDD(Seq("A", "B", "C", "D", "E"), 2)
var rdd3 = sc.makeRDD(Seq("a", "b", "c", "d", "e"), 2)
var rdd4 = rdd1.zipPartitions(rdd2, rdd3) {
  (rdd1Iter, rdd2Iter, rdd3Iter) =>
  {
    var result = List[String]()
    while (rdd1Iter.hasNext && rdd2Iter.hasNext && rdd3Iter.hasNext) {
      result ::= (rdd1Iter.next() + "_" + rdd2Iter.next() + "_" + rdd3Iter.next())
    }
    result.iterator
  }
}
rdd4.foreach(println)


# 輸出結果：
2_B_b
1_A_a
5_E_e
4_D_d
3_C_c

9.zipWithIndex函式與zipWithUniqueId函式

• zipWithIndex函式

該函式將RDD中的元素和這個元素在RDD中的ID（索引號）組合成鍵/值對。

va rdd2 = sc.makeRDD(Seq("A", "B", "R", "D", "F"), 2)
rdd2.zipWithIndex().foreach(println)

# 輸出結果：
(A,0)
(B,1)
(R,2)
(D,3)
(F,4)

• zipWithUniqueId函式

該函式將RDD中元素和一個唯一ID組合成鍵/值對，
該唯一ID生成演算法如下：

每個分割槽中第一個元素的唯一ID值為：該分割槽索引號

每個分割槽中第N個元素的唯一ID值為：(前一個元素的唯一ID值) + (該RDD總的分割槽數)

var rdd1 = sc.makeRDD(Seq("A", "B", "C", "D", "E", "F"), 2)
rdd1.mapPartitionsWithIndex {
  (x, iter) => {
    var result = List[String]()
    while (iter.hasNext) {
      result ::= ("part_" + x + "|" + iter.next())
    }
    result.iterator

  }
}.foreach(println)
rdd1.zipWithUniqueId().foreach(println)

結果如圖：