RDDs之combineByKey()

Posted 2021-01-22 twodoge

combineByKey(crateCombiner,mergeValue,mergeCombiners,partitioner)

最常用的基于Key的聚合函数，返回的类型可以和输入的类型不一样
许多基于key的聚合函数都用到了它，例如说groupByKey()

参数解释

遍历partition中的元素，元素的key，要么之前见过的，要么不是。
如果是新元素，使用我们提供的crateCombiner()函数
如果是这个partition中已经存在的key，则使用mergeValue()函数
合计每个partition的结果的时候，使用mergeCombiners函数

例子1：求平均值

val scores = sc.parallelize(Array(("jake",80.0),("jake",90.0),("jake",85.2),("mike",85.0),("mike",90.0),("mike",78.0)))
求总和
val scores2=scores.combineByKey(score=>(1,score), (c1:(Int, Double), newScore)=>(c1._1+1, c1._2 + newScore), (c1:(Int, Double), c2:(Int, Double))=>(c1._1+c2._1, c1._2+c2._2))
输出：
scores2.foreach(println)
(jake,(3,255.2))
(mike,(3,253.0))
求平均值
val avarage = scores2.map({case(name,(num,score))=>(name,score/num)})
avarage.foreach(println)
(jake,255.2/3)
(mike,253/3)