spark中的RDD算子

aggregateByKey aggregate

scala> def combOp(a :Int, b:Int): Int = {println("combOp:" + a + "\t" + b); a + b }

combOp: (a: Int, b: Int)Int

scala> def seqOP(a : Int, b : Int):Int = {println("seqOp:" + a + "\t" + b); math.min(a, b);}

seqOP: (a: Int, b: Int)Int

scala> val z = sc.parallelize(List(1, 2, 3, 4, 5, 6), 2)

z: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[31] at parallelize at <console>:24

scala> z.aggregate(3)(seqOP,combOp)

seqOp:3 4

seqOp:3 5

seqOp:3 1

seqOp:3 6

seqOp:1 2

seqOp:1 3

combOp:3 3

combOp:6 1

res42: Int = 7

一共分成两个区，初始值为3,定义的seqOP在每个分区的元素间进行聚合，聚合时候初始值为3，，然后用combine函数将每个分区的结果和初始,3，执行combOp操作

aggregateByKey

aggregateByKey中的初始值只需要和reduce函数计算，不需要和combine函数结合计算，输入值为key-value格式，其他的类似。

,reduceByKey reduce

reduce(binary_function)

reduce将RDD中元素前两个传给输入函数，产生一个新的return值，新产生的return值与RDD中下一个元素（第三个元素）组成两个元素，再被传给输入函数，直到最后只有一个值为止。

val c = sc.parallelize(1 to 10)
c.reduce((x, y) => x + y)//结果55

1
2
1
2
[/code]

具体过程，RDD有1 2 3 4 5 6 7 8 9 10个元素，

1+2=3

3+3=6

6+4=10

10+5=15

15+6=21

21+7=28

28+8=36

36+9=45

45+10=55

reduceByKey(binary_function)

reduceByKey就是对元素为KV对的RDD中Key相同的元素的Value进行binary_function的reduce操作，因此，Key相同的多个元素的值被reduce为一个值，然后与原RDD中的Key组成一个新的KV对。

val a = sc.parallelize(List((1,2),(1,3),(3,4),(3,6)))
a.reduceByKey((x,y) => x + y).collect

1
2
1
2
[/code]

//结果 Array((1,5), (3,10))
fold foldByKey

scala> val numbers = List(5, 4, 8, 6, 2)

numbers: List[Int] = List(5, 4, 8, 6, 2)

scala> numbers.fold(0){(z, i) => z + i}

res59: Int = 25

scala> numbers.fold(1){(z, i) => z + i}

res60: Int = 26

fold方法需要输入两个参数：初始值以及一个函数。函数也需要两个参数：累加值和当前的item索引

scala> var rdd1 = sc.makeRDD(Array(("A", 0), ("A", 2), ("B", 1), ("B", 2), ("C", 1)))

rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[45] at makeRDD at <console>:24

scala> rdd1.foldByKey(1)(_+_).collect

res64: Array[(String, Int)] = Array((A,4), (B,5), (C,2))

scala> rdd1.foldByKey(0)(_+_).collect

res65: Array[(String, Int)] = Array((A,2), (B,3), (C,1))

foldByKey按照Key值进行fold

combineByKey

scala> val data = sc.parallelize(List((1, "www"), (1, "iteblog"), (2, "bbs"), (2, "iteblog"), (2, "com"),(3, "good")))

data: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[48] at parallelize at <console>:24

scala> val result = data.combineByKey(List(_), (x:List[String], y:String) => y :: x, (x: List[String], y : List[String])=> x:::y)

result: org.apache.spark.rdd.RDD[(Int, List[String])] = ShuffledRDD[49] at combineByKey at <console>:26

scala> result.collect

res66: Array[(Int, List[String])] = Array((1,List(www, iteblog)), (2,List(bbs, iteblog, com)), (3,List(good)))

对每个相同的key的value执行combine操作。