aggregateByKey(initValue)(seqQp, combineOp [,numberTask]): 是一个shuffle类的算子，与reduceByKey类似，只是它在每个分区seqOp后要应用一次初始值。
代码示例:
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object AggregateByKeyOp {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local").setAppName(AggregateByKeyOp.getClass().getSimpleName)
val sc = new SparkContext(conf)
sc.setLogLevel("ERROR")
val rdd1: RDD[(String, Int)] = sc.parallelize(Array(
("刘备", 1), ("曹操", 2), ("刘备", 3), ("曹操", 4), ("刘备", 100), ("曹操", 200), ("刘备", 300), ("曹操", 400)
), 2)
val rdd2: RDD[String] = rdd1.mapPartitionsWithIndex((index, iter) => {
iter.toList.map(x => "【分区号为：" index ", 值为:" x "】").iterator
})
rdd2.foreach(println)
// 【分区号为：0, 值为:(刘备,1)】
// 【分区号为：0, 值为:(曹操,2)】
// 【分区号为：0, 值为:(刘备,3)】
// 【分区号为：0, 值为:(曹操,4)】
// 【分区号为：1, 值为:(刘备,100)】
// 【分区号为：1, 值为:(曹操,200)】
// 【分区号为：1, 值为:(刘备,300)】
// 【分区号为：1, 值为:(曹操,400)】
val rdd3 = rdd1.aggregateByKey(1000)(_ _, _ _)
rdd3.mapPartitionsWithIndex((index, iter) => {
iter.toList.map(x => "【分区号为：" index ", 值为:" x "】").iterator
}).foreach(println)
// 【分区号为：0, 值为:(曹操,2606)】
// 【分区号为：1, 值为:(刘备,2404)】
rdd3.foreach(println)
// (曹操,2606)
// (刘备,2404)
sc.stop()
}
}
