這篇部落格介紹一下sparkRDD中常用的運算元，首先在叢集上啟動spark-shell：
bin/spark-shell --master spark://shizhan:7077 --total-executor-cores 3 --executor-memory 1g

建立RDD的兩種方式：
a.並行化scala集合：sc.parallelize(Array(1,2,3))
b.從檔案中讀取：sc.textFile("hdfs://shizhan:9000/xxx")

RDD中的運算元分為transformation和action型別，下面是常用運算元介紹：

#檢視該rdd的分割槽數量
rdd1.partitions.length

#collect列印 ---> action
sc.parallelize(Array(1,2,3)).collect

#map,sortBy,fielter ---> transformation
val rdd1 = sc.parallelize(List(5,6,4,7,3,8,2,9,1,10))
val rdd2 = sc.parallelize(List(5,6,4,7,3,8,2,9,1,10)).map(_*2).sortBy(x=>x,true)
val rdd2 = rdd1.map(_*2).sortBy(x=>x+"",true)
val rdd2 = rdd1.map(_*2).sortBy(x=>x.toString,true)
val rdd3 = rdd2.filter(_>10)

#flatMap先map再壓平 ---> transformation
val rdd4 = sc.parallelize(Array("a b c", "d e f", "h i j"))
rdd4.flatMap(_.split(' ')).collect
結果：Array(a, b, c, d, e, f, h, i, j)
val rdd5 = sc.parallelize(List(List("a b c", "a b b"),List("e f g", "a f g"), List("h i j", "a a b")))
rdd5.flatMap(_.flatMap(_.split(" "))).collect
結果：Array(a, b, c, a, b, b, e, f, g, a, f, g, h, i, j, a, a, b)

#union並集 intersection交集 distinct去重 ---> transformation
val rdd6 = sc.parallelize(List(5,6,4,7))
val rdd7 = sc.parallelize(List(1,2,3,4))
val rdd8 = rdd6.union(rdd7)
結果：Array(5, 6, 4, 7, 1, 2, 3, 4)
rdd8.distinct.collect
結果：Array(6, 1, 7, 2, 3, 4, 5)
val rdd9 = rdd6.intersection(rdd7)
結果：Array(4)

#join ---> transformation
val rdd1 = sc.parallelize(List(("tom", 1), ("jerry", 2), ("kitty", 3)))
val rdd2 = sc.parallelize(List(("jerry", 9), ("tom", 8), ("shuke", 7)))
val rdd3 = rdd1.join(rdd2)
結果：Array((tom,(1,8)), (jerry,(2,9)))
val rdd3 = rdd1.leftOuterJoin(rdd2)
結果：Array((tom,(1,Some(8))), (jerry,(2,Some(9))), (kitty,(3,None)))
val rdd3 = rdd1.rightOuterJoin(rdd2)
結果：Array((tom,(Some(1),8)), (jerry,(Some(2),9)), (shuke,(None,7)))

#groupByKey以元組的第一個元素為key --->transformation
val rdd3 = rdd1 union rdd2 ---> Array((tom,1), (jerry,2), (kitty,3), (jerry,9), (tom,8), (shuke,7))
rdd3.groupByKey --->Array[(String, Iterable[Int])] = Array((tom,CompactBuffer(1, 8)), (kitty,CompactBuffer(3)), (jerry,CompactBuffer(2, 9)), (shuke,CompactBuffer(7)))
rdd3.groupByKey.map(x=>(x._1,x._2.sum)) --->Array((tom,9), (kitty,3), (jerry,11), (shuke,7))

#cogroup ---> transformation
val rdd1 = sc.parallelize(List(("tom", 1), ("tom", 2), ("jerry", 3), ("kitty", 2)))
val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("shuke", 2)))
val rdd3 = rdd1.cogroup(rdd2)
結果：Array[(String, (Iterable[Int], Iterable[Int]))] = Array((tom,(CompactBuffer(1, 2),CompactBuffer(1))), (jerry,(CompactBuffer(3),CompactBuffer(2))), (shuke,(CompactBuffer(),CompactBuffer(2))), (kitty,(CompactBuffer(2),CompactBuffer())))
val rdd4 = rdd3.map(t=>(t._1, t._2._1.sum + t._2._2.sum))
結果：Array((tom,4), (jerry,5), (shuke,2), (kitty,2))

#cartesian笛卡爾積 ---> transformation
val rdd1 = sc.parallelize(List("tom", "jerry"))
val rdd2 = sc.parallelize(List("tom", "kitty", "shuke"))
val rdd3 = rdd1.cartesian(rdd2)
結果：Array((tom,tom), (tom,kitty), (tom,shuke), (jerry,tom), (jerry,kitty), (jerry,shuke))

#easy-action
val rdd1 = sc.parallelize(List(1,2,3,4,5), 2)
collect:rdd1.collect
reduce:val rdd2 = rdd1.reduce(_+_)
count:rdd1.count
top:rdd1.top(2)
take:rdd1.take(2)
first(similer to take(1)):rdd1.first
takeOrdered:rdd1.takeOrdered(3)

#mapPartitionsWithIndex把每個partition中的分割槽號和對應的值拿出來 ---> transformation
val func = (index: Int, iter: Iterator[(Int)]) => {
  iter.toList.map(x => "[partID:" +  index + ", val: " + x + "]").iterator
}
val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2)
rdd1.mapPartitionsWithIndex(func).collect
結果：Array([partID:0, val: 1], [partID:0, val: 2], [partID:0, val: 3], [partID:0, val: 4], [partID:1, val: 5], [partID:1, val: 6], [partID:1, val: 7], [partID:1, val: 8], [partID:1, val: 9])

#aggregate 第一個為初始值，每次都要加，第二個為部分操作函式，第三個為全域性操作函式 ---> action
def func1(index: Int, iter: Iterator[(Int)]) : Iterator[String] = {
  iter.toList.map(x => "[partID:" +  index + ", val: " + x + "]").iterator
}
val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2)
rdd1.mapPartitionsWithIndex(func1).collect
結果：Array([partID:0, val: 1], [partID:0, val: 2], [partID:0, val: 3], [partID:0, val: 4], [partID:1, val: 5], [partID:1, val: 6], [partID:1, val: 7], [partID:1, val: 8], [partID:1, val: 9])
rdd1.aggregate(0)(_+_, _+_)
結果：45
rdd1.aggregate(0)(math.max(_, _), _ + _)
結果：13
val rdd2 = sc.parallelize(List("a","b","c","d","e","f"),2)
def func2(index: Int, iter: Iterator[(String)]) : Iterator[String] = {
  iter.toList.map(x => "[partID:" +  index + ", val: " + x + "]").iterator
}
rdd2.mapPartitionsWithIndex(func2).collect
結果：Array[String] = Array([partID:0, val: a], [partID:0, val: b], [partID:0, val: c], [partID:1, val: d], [partID:1, val: e], [partID:1, val: f])
rdd2.aggregate("")(_ + _, _ + _)
結果：String = defabc
rdd2.aggregate("=")(_ + _, _ + _)
結果：String = ==abc=def
val rdd3 = sc.parallelize(List("12","23","345","4567"),2)
rdd3.aggregate("")((x,y) => math.max(x.length, y.length).toString, (x,y) => x + y)
結果：24或42
val rdd4 = sc.parallelize(List("12","23","345",""),2)
rdd4.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y)
結果：10或01
val rdd5 = sc.parallelize(List("12","23","","345"),2)
rdd5.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y)
結果：11

#aggregateByKey ---> transformation
val pairRDD = sc.parallelize(List( ("cat",2), ("cat", 5), ("mouse", 4),("cat", 12), ("dog", 12), ("mouse", 2)), 2)
def func2(index: Int, iter: Iterator[(String, Int)]) : Iterator[String] = {
  iter.toList.map(x => "[partID:" +  index + ", val: " + x + "]").iterator
}
pairRDD.mapPartitionsWithIndex(func2).collect
pairRDD.aggregateByKey(0)(math.max(_, _), _ + _).collect
結果：Array[(String, Int)] = Array((dog,12), (cat,17), (mouse,6))
pairRDD.aggregateByKey(100)(math.max(_, _), _ + _).collect
結果：Array[(String, Int)] = Array((dog,100), (cat,200), (mouse,200))

#coalesce, repartition  --->action
val rdd1 = sc.parallelize(1 to 10, 10)
val rdd2 = rdd1.coalesce(2, false) / val rdd2 = rdd1.repartition(2)
rdd2.partitions.length
結果：2

#collectAsMap --->action
val rdd = sc.parallelize(List(("a", 1), ("b", 2)))
rdd.collectAsMap
結果：Map(b -> 2, a -> 1)

#combineByKey傳入三個函式，分別是初始化函式，部分處理函式，全域性處理函式
val rdd1 = sc.textFile("hdfs://shizhan:9000/wordcount/input/").flatMap(_.split(" ")).map((_, 1))
val rdd2 = rdd1.combineByKey(x=>x,(a:Int,b:Int)=>(a+b),(m:Int,n:Int)=>(m+n))
結果：wordcount
val rdd3 = rdd1.combineByKey(x => x + 10, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)
結果：當input下有3個檔案時(有3個block塊, 不是有3個檔案就有3個block, ), 每個會多加3個10
val rdd4 = sc.parallelize(List("dog","cat","gnu","salmon","rabbit","turkey","wolf","bear","bee"), 3)
val rdd5 = sc.parallelize(List(1,1,2,2,2,1,2,2,2), 3)
val rdd6 = rdd5.zip(rdd4) //zip是transformation
結果：Array[(Int, String)] = Array((1,dog), (1,cat), (2,gnu), (2,salmon), (2,rabbit), (1,turkey), (2,wolf), (2,bear), (2,bee))
val rdd7 = rdd6.combineByKey(List(_), (x: List[String], y: String) => x :+ y, (m: List[String], n: List[String]) => m ++ n)
結果：Array[(Int, List[String])] = Array((1,List(dog, cat, turkey)), (2,List(gnu, wolf, bear, bee, salmon, rabbit)))

#countByKey/Value ---> action
val rdd1 = sc.parallelize(List(("a", 1), ("b", 2), ("b", 2), ("c", 2), ("c", 1)))
rdd1.countByKey
結果：Map(c -> 2, a -> 1, b -> 2)
rdd1.countByValue
結果：Map((b,2) -> 2, (c,2) -> 1, (a,1) -> 1, (c,1) -> 1)

#filterByRange ---> transformation
val rdd1 = sc.parallelize(List(("e", 5), ("c", 3), ("d", 4), ("c", 2), ("a", 1)))
val rdd2 = rdd1.filterByRange("b", "d")
rdd2.collect
結果：Array[(String, Int)] = Array((c,3), (d,4), (c,2))

#flatMapValues ---> transformation
val rdd3 = sc.parallelize(List(("a", "1 2"), ("b", "3 4")))
val rdd4 = rdd3.flatMapValues(_.split(" "))
rdd4.collect
結果：Array((a,1), (a,2), (b,3), (b,4))

#foldByKey ---> transformation
val rdd1 = sc.parallelize(List("dog", "wolf", "cat", "bear"), 2)
val rdd2 = rdd1.map(x => (x.length, x))
結果：Array((3,dog), (4,wolf), (3,cat), (4,bear))
val rdd3 = rdd2.foldByKey("")(_+_)
結果：Array[(Int, String)] = Array((4,bearwolf), (3,dogcat))
val rdd = sc.textFile("hdfs://shizhan:9000/wc").flatMap(_.split(" ")).map((_, 1))
rdd.foldByKey(0)(_+_)
結果：wordcount

#foreachPartition --->action
val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9), 3)
rdd1.foreachPartition(x => println(x.reduce(_ + _)))

keyBy以傳入的引數做key，形成一個個元組 ---> transformation
val rdd1 = sc.parallelize(List("dog", "salmon", "salmon", "rat", "elephant"), 3)
val rdd2 = rdd1.keyBy(_.length)
rdd2.collect
結果：Array[(Int, String)] = Array((3,dog), (6,salmon), (6,salmon), (3,rat), (8,elephant))

#keys values拿出元組的key或者value組成新的元組 --->transformation
val rdd1 = sc.parallelize(List("dog", "tiger", "lion", "cat", "panther", "eagle"), 2)
val rdd2 = rdd1.map(x => (x.length, x))
rdd2.keys.collect
結果：Array(3, 5, 4, 3, 7, 5)
rdd2.values.collect
結果：Array(dog, tiger, lion, cat, panther, eagle)