combineByKey的使用 scala

combineByKey的使用

def combineByKey[C](createCombiner: (V) => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C): RDD[(K, C)]
def combineByKey[C](createCombiner: (V) => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C, numPartitions: Int): RDD[(K, C)]
def combineByKey[C](createCombiner: (V) => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C, partitioner: Partitioner, mapSideCombine: Boolean = true, serializer: Serializer = null): RDD[(K, C)]
其中的参数：
createCombiner：组合器函数，用于将V类型转换成C类型，输入参数为RDD[K,V]中的V,输出为C
mergeValue：合并值函数，将一个C类型和一个V类型值合并成一个C类型，输入参数为(C,V)，输出为C
mergeCombiners：合并组合器函数，用于将两个C类型值合并成一个C类型，输入参数为(C,C)，输出为C
numPartitions：结果RDD分区数，默认保持原有的分区数
partitioner：分区函数,默认为HashPartitioner
mapSideCombine：是否需要在Map端进行combine操作，类似于MapReduce中的combine，默认为true

举例理解：
假设我们要将一堆的各类水果给榨果汁，并且要求果汁只能是纯的，不能有其他品种的水果。那么我们需要一下几步：
1 定义我们需要什么样的果汁。
2 定义一个榨果汁机，即给定水果，就能给出我们定义的果汁。--相当于hadoop中的local combiner
3 定义一个果汁混合器，即能将相同类型的水果果汁给混合起来。--相当于全局进行combiner

那么对比上述三步，combineByKey的三个函数也就是这三个功能
1 createCombiner就是定义了v如何转换为c
2 mergeValue 就是定义了如何给定一个V将其与原来的C合并成新的C
3 就是定义了如何将相同key下的C给合并成一个C
var rdd1 = sc.makeRDD(Array(("A",1),("A",2),("B",1),("B",2),("C",1)))
rdd1.combineByKey(
(v : Int) => List(v),--将1 转换成 list(1)
(c : List[Int], v : Int) => v :: c,--将list(1)和2进行组合从而转换成list(1,2)
(c1 : List[Int], c2 : List[Int]) => c1 ::: c2--将全局相同的key的value进行组合
).collect
res65: Array[(String, List[Int])] = Array((A,List(2, 1)), (B,List(2, 1)), (C,List(1)))
简要介绍

def combineByKey[C](createCombiner: (V) => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C): RD

createCombiner: combineByKey() 会遍历分区中的所有元素，因此每个元素的键要么还没有遇到过，要么就
和之前的某个元素的键相同。如果这是一个新的元素， combineByKey() 会使用一个叫作 createCombiner() 的函数来创建
那个键对应的累加器的初始值
mergeValue: 如果这是一个在处理当前分区之前已经遇到的键，它会使用 mergeValue() 方法将该键的累加器对应的当前值与这个新的值进行合并
mergeCombiners: 由于每个分区都是独立处理的，因此对于同一个键可以有多个累加器。如果有两个或者更
多的分区都有对应同一个键的累加器，就需要使用用户提供的 mergeCombiners() 方法将各
个分区的结果进行合并。

创建一个学生成绩说明的类

case class ScoreDetail(studentName: String, subject: String, score: Float)

下面是一些测试数据，加载测试数据集合 key = Students name and value = https://www.it610.com/article/ScoreDetail instance

val scores = List( ScoreDetail("xiaoming", "Math", 98), ScoreDetail("xiaoming", "English", 88), ScoreDetail("wangwu", "Math", 75), ScoreDetail("wangwu", "English", 78), ScoreDetail("lihua", "Math", 90), ScoreDetail("lihua", "English", 80), ScoreDetail("zhangsan", "Math", 91), ScoreDetail("zhangsan", "English", 80))

换成二元组，也可以理解成转换成一个map, 利用了for 和 yield的组合

val scoresWithKey = for { i <- scores } yield (i.studentName, i) val scoresWithKeyRDD = sc.parallelize(scoresWithKey).partitionBy(new HashPartitioner(3)).cache

聚合求平均值让后打印

val avgScoresRDD = scoresWithKeyRDD.combineByKey( (x: ScoreDetail) => (x.score, 1) /*createCombiner*/, (acc: (Float, Int), x: ScoreDetail) => (acc._1 + x.score, acc._2 + 1) /*mergeValue*/, (acc1: (Float, Int), acc2: (Float, Int)) => (acc1._1 + acc2._1, acc1._2 + acc2._2) /*mergeCombiners*/ // calculate the average ).map( { case(key, value) => (key, value._1/value._2) })avgScoresRDD.collect.foreach(println) /*输出: (zhangsan,85.5) (lihua,85.0) (xiaoming,93.0) (wangwu,76.5) */

解释一下scoresWithKeyRDD.combineByKey
createCombiner: (x: ScoreDetail) => (x.score, 1)
这是第一次遇到zhangsan，创建一个函数，把map中的value转成另外一个类型，这里是把(zhangsan,(ScoreDetail类))转换成(zhangsan,(91,1))
mergeValue: (acc: (Float, Int), x: ScoreDetail) => (acc._1 + x.score, acc._2 + 1) 再次碰到张三，就把这两个合并, 这里是将(zhangsan,(91,1)) 这种类型和 (zhangsan,(ScoreDetail类))这种类型合并，合并成了(zhangsan,(171,2))
mergeCombiners (acc1: (Float, Int), acc2: (Float, Int)) 这个是将多个分区中的zhangsan的数据进行合并，我们这里zhansan在同一个分区，这个地方就没有用上
------------------------------------------------------------------------
contRdd.combineByKey(
(score:(String,Long)) => Map(score._1 -> score._2),
(c:Map[String,Long],score) => (c ++ Map(score._1 -> score._2)),
(c1:Map[String,Long],c2:Map[String,Long]) => (c1 ++ c2) )
【combineByKey的使用】