到網上看了一些資料，簡單的做個筆記。備忘。

測試例子使用的資料：

test01：

a a
b b
c c
d d
e e
f f
g g

1 1
2 2
3 3
4 4
5 5
6 6
a a
b b
c c
d d
e e
f f

1、union(otherRDD)

      union() 將兩個rdd簡單結合在一起，與mysql中的union 操作類似只不過它是操作的rdd，它不會改變partition中的資料

spark sql 測試：

./spark-shell
sc
val t01 = sc.textFile("hdfs://user/data_spark/test01")
val t02 = sc.textFile("hdfs://user/data_spark/test02")
t01.union(t01) foreach println

結果：
a a
e e
b b
a a
f f
b b
c c
c c
g g
d d
d d
e e
f f
g g

 

2、groupByKey(numPartitions)

      普通的RDD 類是沒有這個方法的，org.apache.spark.rdd.PairRDDFunctions 這個pairRdd提供這個方法；

       顧名思義，這個方法是將相同的key的records聚合在一起，類似mysql中的groupby操作，通過ShuffledRDD將每個partition中fetch過來，shuffle機制預設用的是hashShuffle，spark1.1版本引入sorted shuffle，速度更快。shuffle操作後面接著mapPartition（）操作，生成MapPartitionRDD。這就是groupbykey的結果了。

      同一個key的值聚合以後，將所有的value放到一個arraylist，新的arraylist 作為value

val wc = t01.union(t01).flatMap(l=>l.split(" ")).map(w=>(w,1))
wc foreach println
結果：
(e,1)
(e,1)
(e,1)
(e,1)
(f,1)
(f,1)
(f,1)
(f,1)
(g,1)
(g,1)
(g,1)
(g,1)
(a,1)
(a,1)
(a,1)
(a,1)
(b,1)
(b,1)
(b,1)
(b,1)
(c,1)
(c,1)
(c,1)
(c,1)
(d,1)
(d,1)
(d,1)
(d,1)

wc.groupByKey foreach println
結果：
(d,CompactBuffer(1, 1, 1, 1))
(g,CompactBuffer(1, 1, 1, 1))
(c,CompactBuffer(1, 1, 1, 1))
(b,CompactBuffer(1, 1, 1, 1))
(f,CompactBuffer(1, 1, 1, 1))
(e,CompactBuffer(1, 1, 1, 1))
(a,CompactBuffer(1, 1, 1, 1))

 

3、reduceByKey（func，numPartition）

      這個操作的作用類似mapreduce中的reduce操作，對相同的key的值加上func的操作，比如要做wordcount的操作：

             map(x=>(x,1)).reduceByKey(_+_, 5)

接上面的測試

wc.reduceByKey(_+_) foreach println
結果：
(d,4)
(b,4)
(f,4)
(g,4)
(c,4)
(e,4)
(a,4)

4、distinct（numPartitions）

      將 parent rdd 的資料去重，放到新的numPartitions，還是要通過shuffle操作，如果是kv pair 的資料<k,v>  則直接進行shuffle 操作，如果只有key，那麼spark先將資料轉換成<k, null>再進行shuffle。其實後面呼叫的是reduceByKey（）

wc.distinct(1) foreach println
結果：
(g,1)
(b,1)
(f,1)
(d,1)
(a,1)
(e,1)
(c,1)

5、cogroup（otherRDD，numPartitions）

     與groupByKey不同的地方，cogroup 是將多個rdd的資料聚合到一起，過程跟groupByKey 類似.

     但是結果是一個包含多個arraylist 的arraylist，每一個rdd 的value放到一個arraylist，然後，將這些arraylist放到一個元素的arraylist的arraylist。

val wc01 = t01.flatMap(l=>l.split(" ")).map(w=>(w,1))
val wc02 = t02.flatMap(l=>l.split(" ")).map(w=>(w,1))

wc01.cogroup(wc02,1) foreach println
結果：
(d,(CompactBuffer(1, 1),CompactBuffer(1, 1)))
(e,(CompactBuffer(1, 1),CompactBuffer(1, 1)))
(4,(CompactBuffer(),CompactBuffer(1, 1)))
(5,(CompactBuffer(),CompactBuffer(1, 1)))
(a,(CompactBuffer(1, 1),CompactBuffer(1, 1)))
(6,(CompactBuffer(),CompactBuffer(1, 1)))
(b,(CompactBuffer(1, 1),CompactBuffer(1, 1)))
(2,(CompactBuffer(),CompactBuffer(1, 1)))
(3,(CompactBuffer(),CompactBuffer(1, 1)))
(f,(CompactBuffer(1, 1),CompactBuffer(1, 1)))
(1,(CompactBuffer(),CompactBuffer(1, 1)))
(g,(CompactBuffer(1, 1),CompactBuffer()))
(c,(CompactBuffer(1, 1),CompactBuffer(1, 1)))

6、intersection(otherRDD)

     這個操作值保留兩個rdd中都包含的資料，首先將rdd的資料轉化成<k, ->，後面呼叫cogroup（）操作。

     然後，  對cogroup結果進行過濾，由前面cogroup 的結果格式介紹可知，會生成包含兩個arraylist元素的arraylist，只保留結果中兩個arraylist都不為空的，最後取出key，便是最終的結果。

wc01.intersection(wc02) foreach println
結果：
(d,1)
(e,1)
(b,1)
(f,1)
(a,1)
(c,1)

只有兩個rdd共同的部分 kv 對

7、join（otherRDD， numPartitions）

     將兩個RDD[ K, V ] 安裝sql中的join方式聚合。類似intersection，先進行cogroup操作，得到<k, (Iterable[v1], Iterable[v2])> 的MappedValuesRDD。

     將 Iterable[v1] 和 Iterable[v2] 做笛卡爾集，並將集合flat（）化，生成FlatMappedValuesRDD。

wc01.join(wc02,1) foreach println
結果：
(d,(1,1))
(d,(1,1))
(d,(1,1))
(d,(1,1))
(e,(1,1))
(e,(1,1))
(e,(1,1))
(e,(1,1))
(a,(1,1))
(a,(1,1))
(a,(1,1))
(a,(1,1))
(b,(1,1))
(b,(1,1))
(b,(1,1))
(b,(1,1))
(f,(1,1))
(f,(1,1))
(f,(1,1))
(f,(1,1))
(c,(1,1))
(c,(1,1))
(c,(1,1))
(c,(1,1))

8、sortByKey（ascending，numPartitions）

     將RDD [ k, v ] 按照key進行排序，如果ascending=true表示升序，false表示降序。

     先通過shuffle將資料聚合到一起，然後將聚合的資料按照key排序

wc01.sortByKey(true,1) foreach println
結果：
(a,1)
(a,1)
(b,1)
(b,1)
(c,1)
(c,1)
(d,1)
(d,1)
(e,1)
(e,1)
(f,1)
(f,1)
(g,1)
(g,1)

9、cartesian（otherRDD）

       求兩個rdd的笛卡爾集，生成的CartesianRDD中的partition個數為兩個rdd的partition的個數乘積。

       邏輯類似join

笛卡爾乘積，這個很簡單，不過資料量大的話就不要這麼做了

10、coalesce（numPartitions， shuffle = false）

        合併，對一個rdd，兩種方式，一種需要shuffle，一種直接將多個partitions的內容合併到一起，不需要shuffle。

        這個方法的主要作用就是調整 parentRDD 的partition數量。合併因素除了考慮partition的個數外，還應該考慮locality 和 balance的問題

這個操作的邏輯比較難理解：

 wc01.coalesce(1) foreach println
結果：
14/10/27 17:23:42 INFO rdd.HadoopRDD: Input split: hdfs://qunarcluster/user/data_spark/test01:0+14
(a,1)
(a,1)
(b,1)
(b,1)
(c,1)
(c,1)
(d,1)
(d,1)
14/10/27 17:23:42 INFO rdd.HadoopRDD: Input split: hdfs://qunarcluster/user/data_spark/test01:14+14
(e,1)
(e,1)
(f,1)
(f,1)
(g,1)
(g,1)

 wc01.coalesce(2) foreach println
結果：
(a,1)
(a,1)
(b,1)
(b,1)
(c,1)
(c,1)
(e,1)
(d,1)
(e,1)
(d,1)
(f,1)
(f,1)
(g,1)
(g,1)

11、repartition（numPartitions）       

        等價於coalesce(numPartitions, shuffle = true)

wc01.repartition(1) foreach println
(a,1)
(a,1)
(b,1)
(b,1)
(c,1)
(c,1)
(d,1)
(d,1)
(e,1)
(e,1)
(f,1)
(f,1)
(g,1)
(g,1)

雖然mapreduce 相當於 spark 的 map +  reduceByKey， 但是 mapreduce中的reduce可以靈活的操作，加入一些自己的邏輯，所以，各有所長。

但是，spark 確實很方便