Spark|KeyValue对RDDs之combineByKey函数
一 combineByKey() (createCombiner,mergeValue,mergeCombiners,partitioner) 最常用的基于key的聚合函数，返回的类型可以与输入类型不一样 许多基于key的聚合函数都用到了它，向groupByKey()
二 combineByKey工作原理 遍历parition中的元素，元素的key，要么之前见过的，要么没见过。 如果是新元素，使用我们提供的creatCombiner()函数，相对于初始化 如果是这个parition中已经存在的key，就会使用mergeValue()函数 合计每个partition的结果的时候，使用mergeCominers()函数
三 实战：求平均值
scala> val scores=sc.parallelize(Array(("jake",80.0),("jake",90.0),("jake",85.0),("mike",85.0),("mike",93.0),("mike",89.0)))
scores: org.apache.spark.rdd.RDD[(String, Double)] = ParallelCollectionRDD[10] at parallelize at :24
scala> scores.foreach(println)
(jake,80.0)
(jake,90.0)
(jake,85.0)
(mike,85.0)
(mike,93.0)
(mike,89.0)
scala> val score2=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))
score2: org.apache.spark.rdd.RDD[(String, (Int, Double))] = ShuffledRDD[11] at combineByKey at :26
scala> score2.foreach(println)
(jake,(3,255.0))
(mike,(3,267.0))
scala> val areage=score2.map{case(name,(num,score))=>(name,score/num)}
areage: org.apache.spark.rdd.RDD[(String, Double)] = MapPartitionsRDD[12] at map at :28
scala> areage.foreach(println)
(jake,85.0)
(mike,89.0) 推荐阅读
- 布丽吉特，人生绝对的赢家
- 进必趋|进必趋 退必迟，问起对 视勿移
- 对称加密和非对称加密的区别
- 对抗抑郁最好的方法
- 装聋作哑，关系融洽
- 社保代缴公司服务费包含哪些
- 数组常用方法一
- 幸福的婚姻不争对错！读《你要的是幸福还是对错》有感。
- 把一切献给现在
- 年轻人对未来迷茫时该怎么办（）