从saprk的wordcount解析groupBykey与RudeceByKey的源码
1.引出问题
有两个wordcount的书写方式：
1.这是个官方教程的书写wordcount
scala> sc.parallelize(List(List("who am i"),List("where you want to go","i will go with you")))
res21: org.apache.spark.rdd.RDD[List[String]] = ParallelCollectionRDD[17] at parallelize at <console>:25
scala> res21.flatMap(_.flatMap(_.split(" "))).map((_,1)).reduceByKey(_+_).collect
res22: Array[(String, Int)] = Array((am,1), (with,1), (will,1), (who,1), (go,2), (want,1), (you,2), (i,2), (to,1), (where,1))
2.下面一种是学习scala的时候学习的一种wordcount书写方式
scala> res21.flatMap(_.flatMap(_.split(" "))).map((_,1)).groupByKey.map(t=>(t._1,t._2.sum)).collect
res24: Array[(String, Int)] = Array((am,1), (with,1), (will,1), (who,1), (go,2), (want,1), (you,2), (i,2), (to,1), (where,1))
我发现两种方式都能够得到wordcount的结果，于是我就想哪一种速率比较快？
2.进行比较
2.1 groupByKey函数
scala> val rdd1 = sc.parallelize(List(("tom", 1), ("jerry", 3), ("kitty", 2)))
rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[6] at parallelize at <console>:24
scala> val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("shuke", 2)))
rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[7] at parallelize at <console>:24
scala> val rdd3 = rdd1 union rdd2
rdd3: org.apache.spark.rdd.RDD[(String, Int)] = UnionRDD[5] at union at <console>:28
scala> rdd3.groupByKey.collect
res1: Array[(String, Iterable[Int])] = Array((tom,CompactBuffer(1, 1)), (jerry,CompactBuffer(2, 3)), (shuke,CompactBuffer(2)), (kitty,CompactBuffer(2)))
实验发现，groupByKey函数会直接聚合结果集：
1.创建聚合器
2.将k相同的v编程compactBuffer聚合成为一个元素为compactBuffer的集合
3.调用combineByKeyWithClassTag聚合起来
查看源码：根据key聚合所有的value在一个sequence中
下面补充说明了：If you are grouping in order to perform an aggregation (such as a sum or average) over each key, using PairRDDFunctions.aggregateByKey or PairRDDFunctions.reduceByKey will provide much better performance.表示聚合操作和reduce操作，使用PairRDDFunctions.aggregateByKey和PairRDDFunctions.aggregateByKey更好！
注意:可以看到groupByKey中的combineByKeyWithClassTag中的mapSideCombine = false，表示在map端聚合的操作被禁止了，这就是groupByKey比 reduceByKey慢的原因
2.2 reduceByKey
reduce调用了combineByKeyWithClassTag函数实现聚合操作，将多个partition中的数据聚合在一起。
2.3 aggregateByKey
很明显aggregateByKey的操作就是：
1.赋初值
2.获取saprk环境，并序列化
3.调用combineByKeyWithClassTag
2.4 上面三个函数都调用了combineByKeyWithClassTag
def combineByKeyWithClassTag[C](
createCombiner: V => C,
mergeValue: (C, V) => C,
mergeCombiners: (C, C) => C,
partitioner: Partitioner,
mapSideCombine: Boolean = true,
serializer: Serializer = null)(implicit ct: ClassTag[C]): RDD[(K, C)] = self.withScope
参数说明：
- createCombiner：map端，改变 v 的返回值类型为c，类似于初始化操作
- mergeValue：在每个分区中的操作，类似于mapreduce中的map
- mergeCombiners：分区聚合后的操作，类似于mapeduce中的reduce
- partitioner：分区对象
- mapSideCombine：默认开启map端的聚合操作
- serializer：序列化
3.分析
3.1 reduceByKey( _ + _ )和aggregateByKey( _ + _ , _ +_ )
reduceByKey( _ + _ )和aggregateByKey( _ + _ , _ + _ )的本质是相同的：
scala> val pairRDD = sc.parallelize(List( ("cat",2), ("cat", 5), ("mouse", 4),("cat", 12), ("dog", 12), ("mouse", 2)), 2)
pairRDD: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[5] at parallelize at <console>:24
scala> pairRDD.aggregateByKey(0)(_ + _, _ + _).collect
res19: Array[(String, Int)] = Array((dog,12), (cat,19), (mouse,6))
scala> pairRDD.reduceByKey( _ + _).collect
res20: Array[(String, Int)] = Array((dog,12), (cat,19), (mouse,6))
他们会先在每个patiition中map求和，然后再将每个partition的结果reduce求和。
3.2 reduceByKey( _ + _ )和groupByKey()
方式1：
scala> res21.flatMap(_.flatMap(_.split(" "))).map((_,1)).reduceByKey(_+_).collect
方式2：
scala> res21.flatMap(_.flatMap(_.split(" "))).map((_,1)).groupByKey.map(t=>(t._1,t._2.sum)).collect
方式1之所以比方式2快是因为，reduce会先在每个partition中求和（并行），然后在对每个partition中的结果求和，而groupByKey.map(t=>(t._1,t._2.sum))会先聚合所有的pairRDD，然后在一个patition中对所有数据求和。原因在于groupByKey禁止了mapSideCombine。
