本文要介绍的是Spark Core中的Aggregator这个类。这个类的用处非常大，为什么这么说呢？我们都知道Spark支持传统的MapReduce模型，并基于这种模型提供了比Hadoop更多更高层次的计算接口。比如Spark Core PairRDD中非常常用的：

• reduceByKey  提供聚合函数，将k-v对集合将相同key值的value聚合，这个方法会先在map端执行减少shuffle量，然后在reduce端执行
• aggregateByKey 与reduceByKey类似，不过会将k-v对集合聚合变形为新的k-u类型的RDD。需要提供两个方法：seqOp[(U,V)=>U]在单个分区内将原始V类型的值merge到U类型的汇总值方法，以及combineOp[(U,U)=>]在不同分区间将聚合结果merge的方法。
• groupByKey 将原本的RDD，根据key值进行分组，返回RDD[K, Iterable[V]]结果。顺带说一嘴：这个方法不会做map端的combine操作（因为实际上数据结果并没有减少，反而因为需要插入到hash table中会增加老年代的内存压力）
• ……

而这几种方法底层都是依赖于combineByKeyWithClassTag（在Spark1.4中是combineByKey，新版本接口增加了ClassTag，支持原生类型）。这个函数的实现方式如以下代码所示：

该函数依赖于六个参数，计算逻辑如图

1. createCombiner: V => C 提供聚合过程的初始值的函数
2. mergeValue: (C, V) => C 将V值merge到聚合值中的方法
3. mergeCombiners: (C, C) => C 将聚合值进行聚合的方法
4. partitioner: Partitioner 指定分区方法
5. mapSideCombine: Boolean = true 告诉ShuffleRDD（其实是告诉它依赖的ShuffleDependency）是否需要做Map端的combine 过程
6. serializer: Serializer = null 序列化，一般不用指定

如图可以看出来，由于Spark懒式计算的原则，现在只是生成了MapPartitionsRDD 或者 ShuffledRDD，当遇到类似collect / count 等操作的时候这些RDD就会依赖DAGScheduler 的调度，递归的先执行依赖，然后一步步执行完成。当然，我们的主角Aggregator 在计算Shuffled Dependency的时候就会完全发挥出作用来了。具体它是怎么做的呢？

我们都知道Shuffle RDD 的行为是新创建一个stage，然后顺次执行它依赖的stage，并且读取最终执行完的结果。而从上面我们可以看到，我们将Aggregator 设置到了Shuffle Dependency 中了，所以我们猜测这个聚合操作就应该在读取执行结果的前后过程中。果然，在HashShuffleWriter 和SortShuffleWriter（后者用的更多些）的write方法中，在BlockStoreShuffleReader 这个类的read 方法中，我们找到了：

在这里我们看到了，通过调用Aggregator的以下两个方法就完成了Shuffle过程中map端和reduce端的聚合操作：

1. def combineValuesByKey(iter: Iterator[_ <: Product2[K, V]], context: TaskContext):Iterator[(K, C)] 直接对依赖RDD 的值进行聚合
2. def combineCombinersByKey(iter: Iterator[_ <: Product2[K, C]], context: TaskContext):Iterator[(K, C)] 对依赖RDD 产出的聚合结果进行再次聚合

这两个函数的逻辑非常相似，我只把第一个函数贴出来分析里面进行的操作

我们看到，这里Spark使用了一个ExternalAppendOnlyMap 这样的Map 结构，将所有的数据insertAll 到这个结构里面来，结果调用iterator 方法就可以获得最终结果了。我们可以理解，如果是一个Map 的话，我们可以将所有的key 作为map 的key，然后迭代取上一轮的结果，判断key 如果存在于map 中的话，将value 取出，跟新的数据做聚合，然后写回map中即可。原来Aggregator 的实现也很简单嘛，跟我们单机版的并没有什么不同。但现在有一个最重要的问题：如果数据量太大，内存中存储不下了怎么办？

这种情况在大数据场景下几乎是必然发生的问题。别担心，ExternalAppendOnlyMap 这个名字中的External 就告诉我们它肯定是可以向磁盘进行Spill 的，它也是一个Spark Spill 到磁盘的机制很典型的例子。我们来看看它是如何优雅的处理这个问题的。

如左图，我们可以看出ExternalAppendOnlyMap 持续地读取数据，然后在SizeTrackingAppendOnlyMap （使用开放寻址和二次探测的方式实现的不可删除的HashMap，为什么要这么实现？留作一个问题，大家请自己思考。）中不断做聚合操作。如果数据一旦超出规定的阈值，就将currentMap按照hash 值排序后spill 到磁盘上（按照hash 排序很重要！！！），然后创建一个新的map继续重复这样的操作。

但大家就会问了，这样子的话，一个key 可能会存在在内存中、多个DiskIterator 中，那其实并没有完成真正的数据聚合啊？这时候，当insertAll 完成之后，我们将会调用iterator 方法，在这里是真正完成聚合的关键所在（右图所示）。iterator 返回了一个基于内存中currentMap 和DiskIterator 两部分数据的多路归并迭代器。这个迭代器，每次在调用next 方法的时候都会在内部的优先级队列（按每个迭代器最小hash值作为比较对象的堆结构），寻找最小的hash值且key值相等的所有元素（因为我们每个map 都是排序过的，所以这总能实现），进行merge，将所有符合要求的元素merge完成后返回。这样便完成了最终的聚合操作。

所以我们总结下：

1. groupByKey/ reduceByKey/ aggregateByKey/ foldByKey 等都是用Aggregator 实现的。其中groupByKey 没有做Map 端的combine，且分组操作比较重，如果只是要做聚合操作，那建议用后三种操作。
2. Aggregator 作用的位置是ShuffleWriter 和ShuffleReader 的write 和read 过程，分别完成map 端的combine 和 reduce端的聚合。
3. Aggregator 内部实现考虑到了内存不足，进行磁盘spill 的场景，它采用多个基于开放寻址的不可删除SizeTrackingAppendOnlyMap 进行聚合，然后内存超过阈值是进行spill，最后迭代器中多路归并完成聚合操作。

对我们的优化的意义：

1. 不要进行groupByKey.map(_._2.size) 类似这样的操作来统计每个key的count数，因为groupByKey 操作非常重，这种情况用其它聚合方式
2. Aggregator 使用的还是基于近于java的HashMap的方式进行内存中的聚合的，这个方式是比较消耗内存的，所以在这种过程中很容易发生多次磁盘Spill，容易在老年代生成很多对象，容易发生GC，导致性能问题。所以在这种情况下就要求
   1. 合理进行分区，要对自己数据进行更多的测试，分区数量要足够，否则很容易出现性能问题
   2. 如果Shuffle数据量太大的话，建议不要使用这种方式，可以使用repartitionAndSortWithinPartition这种函数做特异性优化。