Spark聚合函数-aggregateByKey是如何工作的？

Question

**row   | id,  c.**
---------------
row1  | k1 , c1  
row2  | k1 , c2  
row3  | k1 , c3  
row4  | k2 , c4  
row5  | k2 , c5  
row6  | k2 , c6  
row7  | k3 , c7  
row8  | k3 , c8  
row9  | k3 , c9  
row10 | k4 , c10   
row11 | k4 , c11  
row12 | k4 , c12 

null

Answer 1

aggregateByKey（）与ReduceByKey有很大不同。发生的情况是reduceByKey是AggregateByKey的一种特殊情况。

aggregateByKey（）将组合特定键的值，这种组合的结果可以是您指定的任何对象。您必须指定如何在一个分区（在同一节点中执行）内组合（“Added”）值，以及如何组合来自不同分区（可能在不同节点中）的结果。reduceByKey是一个特例，因为组合的结果（例如和）与值的类型相同，从不同分区组合的操作也与在分区内组合值的操作相同。

举个例子：假设你有一个对的列表。你把它并行化：

val pairs = sc.parallelize(Array(("a", 3), ("a", 1), ("b", 7), ("a", 5)))

val resReduce = pairs.reduceByKey(_ + _) //the same operation for everything
resReduce.collect
res3: Array[(String, Int)] = Array((b,7), (a,9))

//0 is initial value, _+_ inside partition, _+_ between partitions
val resAgg = pairs.aggregateByKey(0)(_+_,_+_)
resAgg.collect
res4: Array[(String, Int)] = Array((b,7), (a,9))

import scala.collection.mutable.HashSet
//the initial value is a void Set. Adding an element to a set is the first
//_+_ Join two sets is the  _++_
val sets = pairs.aggregateByKey(new HashSet[Int])(_+_, _++_)
sets.collect
res5: Array[(String, scala.collection.mutable.HashSet[Int])]  =Array((b,Set(7)), (a,Set(1, 5, 3)))