combineByKey、reduceByKey、groupBy

combineByKey

• combineByKey实际上是将参数传递给了combineByKeyWithClassTag来完成工作的

• reduceByKey和groupByKey底层也都是调用了combineByKeyWithClassTag这个方法

• combineByKeyWithClassTag

  • 针对pari RDD(k,v)进行操作， 使用自定义的聚合函数对相同key的元素进行聚合
  • 将(k,v)类型的数据转换成(k,c)类型的数据

  def combineByKeyWithClassTag[C](
      createCombiner: V => C,
      mergeValue: (C, V) => C,
      mergeCombiners: (C, C) => C,
      partitioner: Partitioner,
      mapSideCombine: Boolean = true,
      serializer: Serializer = null)(implicit ct: ClassTag[C]): RDD[(K, C)] = self.withScope {...}

• 必须要传三个参数

  • createCombiner：当第一次遇到key时，调用这个函数，将key对应的V转换成C（初始化操作）
  • mergeValue：不是第一次遇到key时，调用这个函数，将key对应的V累加到第一次的C中（对这个分区中相同的key的进一步操作）
  • mergeCombiners：针对不同分区的操作，将相同Key的C合并成一个C

• 默认参数

  • partitioner：分区函数,默认为HashPartitioner
  • mapSideCombine：是否需要在Map端进行combine操作，类似于MapReduce中的combine，默认为true
  • serializer：序列化，用于数据存储和传输

• 返回值RDD[(K, C)]

  • 表示根据相同的k，将value值由原来的V类型最后转换为C类型

案例一：

val a = sc.parallelize(List("dog","cat","gnu","salmon","rabbit","turkey","wolf","bear","bee"), 3)
val b = sc.parallelize(List(1,1,2,2,2,1,2,2,2), 3)
val c = b.zip(a)        // 利用拉链操作将两个rdd转换为pari rdd

val d = c.combineByKey(
    List(_),            // 取出第一次出现的key的val，加入到list中
    (x:List[String], y:String) => y :: x,       // 取出第二次出现的key的val，加入到第一次key对应的list中
    (x:List[String], y:List[String]) => x ::: y // 将不同分区，相同的key，对应的list连接到一起
)


d.collect
res: Array[(Int, List[String])] = Array((1,List(cat, dog, turkey)), (2,List(gnu, rabbit, salmon, bee, bear, wolf)))

案例二：

val initialScores = Array(("Fred", 88.0), ("Fred", 95.0), ("Fred", 91.0), ("Wilma", 93.0), ("Wilma", 95.0), ("Wilma", 98.0))  
val d1 = sc.parallelize(initialScores)

// 定义一个元组类型(科目计数器,分数)
// type的意思是以后再这个代码中所有的类型为(Int, Double)都可以被记为MVType
type MVType = (Int, Double) 
d1.combineByKey(  
  score => (1, score),  // 以分数作为参数,对分数进行标记
  // 注意这里的c1就是createCombiner初始化得到的结果(1, score)，对相同的key进行标记+1，分数累加
  (c1: MVType, newScore) => (c1._1 + 1, c1._2 + newScore),  
  // 对不同分区，相同key的(n, score)进行累加
  (c1: MVType, c2: MVType) => (c1._1 + c2._1, c1._2 + c2._2)  
).map 
{ 
case (name, (num, socre)) 
=> (name, socre / num)
 }.collect 

reduceByKey

• reduceByKey底层也是通过combineByKeyWithClassTag来实现的

reduceByKey的源码

• combineByKeyWithClassTag的第一个参数默认为(v: V) => v，所以对元素不会产生任何影响
• 第二、三两个参数都一样，是reduceByKey传递过来的，将两个值变成一个值(V, V) => V

def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)] = self.withScope {
combineByKeyWithClassTag[V]((v: V) => v, func, func, partitioner)
}

案例:

val a = sc.parallelize(List("dog", "tiger", "lion", "cat", "panther", "eagle"), 2)
val b = a.map(x => (x.length, x))    // 生成一个键值对类型的数据(字符串长度，字符串)
b.reduceByKey(_ + _).collect         // 将字符串长度相同的key，所对应的val累加
res: Array[(Int, String)] = Array((4,lion), (3,dogcat), (7,panther), (5,tigereagle))

groupByKey

• groupByKey底层也是通过combineByKeyWithClassTag来实现的

groupByKey 源码

• groupByKey的返回值为RDD[(K, Iterable[V])]，val值是一个迭代器，其内容包含所有key值为K的元祖的value值
• 执行过程类似reduceByKey，只是已经帮你写好了每个函数，但是参数mapSideCombine = false，也就意味着，不在map端执行，在reduce端执行
• 所以在大量数据处理的情况下：groupByKey不如reduceByKey

def groupByKey(partitioner: Partitioner): RDD[(K, Iterable[V])] = self.withScope {
val createCombiner = (v: V) => CompactBuffer(v)
val mergeValue = (buf: CompactBuffer[V], v: V) => buf += v
val mergeCombiners = (c1: CompactBuffer[V], c2: CompactBuffer[V]) => c1 ++= c2
val bufs = combineByKeyWithClassTag[CompactBuffer[V]](
  createCombiner, mergeValue, mergeCombiners, partitioner, mapSideCombine = false)
bufs.asInstanceOf[RDD[(K, Iterable[V])]]
}

案例：

val a = sc.parallelize(List("dog", "tiger", "lion", "cat", "spider", "eagle"), 2)
//keyBy算子的意思是以_.length这个值作为key，其中value的返回值为ArrayBuffer。
val b = a.keyBy(_.length) 
b.groupByKey.collect

res: Array[(Int, Seq[String])] = Array((4,ArrayBuffer(lion)), (6,ArrayBuffer(spider)), (3,ArrayBuffer(dog, cat)), (5,ArrayBuffer(tiger, eagle)))