當資料集一鍵值對形式組織的時候，聚合具有相同鍵的元素進行一些統計是很常見的操作。對於Pair RDD常見的聚合操作如：reduceByKey，foldByKey，groupByKey，combineByKey。這裡重點要說的是combineByKey。因為combineByKey是Spark中一個比較核心的高階函式，groupByKey,reduceByKey都是基於combineByKey實現的。

combineByKey的定義

def combineByKey[C](  
      createCombiner: V => C,  
      mergeValue: (C, V) => C,  
      mergeCombiners: (C, C) => C,  
      partitioner: Partitioner,  
      mapSideCombine: Boolean = true,  
      serializer: Serializer = null)  
 

其中的引數：

createCombiner: V => C ，這個函式把當前的值作為引數，此時我們可以對其做些附加操作(型別轉換)並把它返回 (這一步類似於初始化操作)

mergeValue: (C, V) => C，該函式把元素V合併到之前的元素C(createCombiner)上 (這個操作在每個分割槽內進行)

mergeCombiners: (C, C) => C，該函式把2個元素C合併 (這個操作在不同分割槽間進行)

numPartitions：結果RDD分割槽數，預設保持原有的分割槽數
partitioner：分割槽函式,預設為HashPartitioner
mapSideCombine：是否需要在Map端進行combine操作，類似於MapReduce中的combine，預設為true

combineByKey來求解平均數的例子

val initialScores = Array(("Fred", 88.0), ("Fred", 95.0), ("Fred", 91.0), ("Wilma", 93.0), ("Wilma", 95.0), ("Wilma", 98.0))  
val d1 = sc.parallelize(initialScores)  
type MVType = (Int, Double) //定義一個元組型別(科目計數器,分數)  
d1.combineByKey(  
  score => (1, score),  
  (c1: MVType, newScore) => (c1._1 + 1, c1._2 + newScore),  
  (c1: MVType, c2: MVType) => (c1._1 + c2._1, c1._2 + c2._2)  
).map { case (name, (num, socre)) => (name, socre / num) }.collect  
 

引數含義的解釋
a 、score => (1, score)，我們把分數作為引數,並返回了附加的元組型別。 以"Fred"為列，當前其分數為88.0 =>(1,88.0) 1表示當前科目的計數器，此時只有一個科目
b、(c1: MVType, newScore) => (c1._1 + 1, c1._2 + newScore)，注意這裡的c1就是createCombiner初始化得到的(1,88.0)。在一個分割槽內，我們又碰到了"Fred"的一個新的分數91.0。當然我們要把之前的科目分數和當前的分數加起來即c1._2 + newScore,然後把科目計算器加1即c1._1 + 1
c、 (c1: MVType, c2: MVType) => (c1._1 + c2._1, c1._2 + c2._2)，注意"Fred"可能是個學霸,他選修的科目可能過多而分散在不同的分割槽中。所有的分割槽都進行mergeValue後,接下來就是對分割槽間進行合併了,分割槽間科目數和科目數相加分數和分數相加就得到了總分和總科目數