1. 前言

有时候需要按照key去做reduce操作时，一般情况下调用reduceByKey就可以完成按照key reduce的任务，reduceByKey的调用就必然意味着shuffle操作。但是有的时候如果我们已经知道相同的key都在同一个partition里面了，这个时候其实没有必要去使用reduceByKey通过一次shuffle将相同的key收集到同一个reducer分区里面，而是可以直接在map端就去完成reduce操作。

比如下面是一个word count在2个分区里面的分布：

------partition 1----------
(failure,1)
(count,1)
(thief,1)
(failure,1)

------partition 2--------
(fortification,1)
(peek,1) 
(lepta,1)
(peek,1)

由于相同的word都在同一个分区里面了，没必要去通过reduceByKey去完成word count操作。

2. 解决方法

实现一个RDD，在其compute方法里完成按key聚合，实现如下:

/**
 K: key type
 V: 上游rdd中value的type
 C: V 经过reduce之后的type
参考ShufferedRDD
*/
class MapsideReduceRDD[K:ClassTag, V:ClassTag, C:ClassTag](
   // 上游rdd，要求上游的rdd中数据已经转换成(key,value)的形式
  @transient var prev : RDD[_ <: Product2[K,V]]

) extends RDD[(K,C)](prev){
  // 需要一个aggregator去完成value的聚合, reduceByKey也会创建这个
  private var aggregator : Option[Aggregator[K,V,C]] = None

  def setAggregator(aggregator: Aggregator[K,V,C]):this.type ={
    this.aggregator = Option(aggregator)
    this
  }

  override def compute(split: Partition, context: TaskContext): Iterator[(K, C)] = {
    /* 创建一个ExternalAppendOnlyMap，这个数据结构是spark中提供的，插入
 .  (K,V)的数据，然后按照传递给他聚合方法完成(K,V)的聚合，返回(K,C)的数据
   */
    val externalMap = createExternalMap
  // 这里迭代上游rdd中(K,V)类型的记录
    val rddIter = dependencies(0).rdd.asInstanceOf[RDD[Product2[K,V]]].iterator(split, context)
    // 插入到externalMap中
    externalMap.insertAll(rddIter)
   // 返回
    new InterruptibleIterator(context,
      externalMap.iterator
    )
  }

  override protected def getPartitions: Array[Partition] = firstParent[Product2[K,V]].partitions

  private def createExternalMap: ExternalAppendOnlyMap[K,V,C] = {
    require(aggregator.nonEmpty, "aggregator should not be empty")
    /**
      创建ExternalAppendOnlyMap， 它需要一下参数:
      - 一个V => C 类型的函数，用于迭代时发现某个key的第一个value，将它转换成C
     - 一个(C,V) => C类型的函数，用于将value合并到C上
     - 一个(C,C) => C类型的函数，将两个部分聚合的结果合并到一起
     */
    new ExternalAppendOnlyMap[K,V,C](aggregator.get.createCombiner, aggregator.get.mergeValue, aggregator.get.mergeCombiners)
  }
}

ExternalAppendOnlyMap会在必要时spill到磁盘

2.1 测试

测试类如下：

object MapsideReduceTest {
  def main(args: Array[String]): Unit ={
    val sc = new SparkContext()
    val words = Seq(Seq("failure","count","thief","failure","count"),Seq("fortification","peek","lepta","peek"));
    // 分两个分区，第一个分区包含Seq("failure","count","thief","failure","count")
    // 这样相同的word只在一个分区里面，然后统计word count
    val wordsRDD = sc.parallelize(words, 2)
    // flatMap将Seq()展开，然后调用map转换成(failure,1)这种数据
    val wordsCount = wordsRDD.flatMap(seq => seq).map(word => (word,1))

    val aggregator = createAggregator
    val mapsideReduceRDD = new MapsideReduceRDD[String, Int, Int](wordsCount).setAggregator(aggregator)
    mapsideReduceRDD.saveAsTextFile("/Users/eric/mapsideReduce")
  }

  def createAggregator: Aggregator[String,Int,Int] ={
    val createCombiner: Int => Int = value => value
    val mergeValue : (Int, Int) => Int = (mergedValue, newValue) => {
      mergedValue + newValue
    }
    val mergeCombiner = mergeValue

    new Aggregator[String, Int,Int](createCombiner, mergeValue, mergeCombiner)
  }
}

提交后测试结果如下：

产生2个输出文件以及内容：
------ part-00000 -------
(failure,2)
(count,2)
(thief,1)

------- part-00001 ------
(lepta,1)
(peek,2)
(fortification,1)