编程

当前位置:时时彩平台 > 编程 > spark RDD,reduceByKey vs groupByKey

spark RDD,reduceByKey vs groupByKey

来源:http://www.mrmtshipyard.com 作者:时时彩平台 时间:2019-10-05 14:27

Spark 中有两个类似的api,分别是 reduceByKey 和 groupByKey 。这两个的功能类似,但底层实现却有些不同,那么为什么要这样设计呢?我们来从源码的角度分析一下。

键值对的RDD操作与基本RDD操作一样,只是操作的元素由基本类型改为二元组。

先看两者的调用顺序(都是使用默认的Partitioner,即defaultPartitioner)

 

 

所用 spark 版本:spark 2.1.0

概述

键值对RDD是Spark操作中最常用的RDD,它是很多程序的构成要素,因为他们提供了并行操作各个键或跨界点重新进行数据分组的操作接口。

 

 

Step1

创建

Spark中有许多中创建键值对RDD的方式,其中包括

  • 文件读取时直接返回键值对RDD
  • 通过List创建键值对RDD

在Scala中,可通过Map函数生成二元组

val listRDD = sc.parallelize(List(1,2,3,4,5))
val result = listRDD.map(x => (x,1))
result.foreach(println)

//结果
(1,1)
(2,1)
(3,1)
(4,1)
(5,1)

 

 

 def reduceByKey(func:  => V): RDD[] = self.withScope { reduceByKey(defaultPartitioner, func) }

键值对RDD的转化操作

 

基本RDD转化操作在此同样适用。但因为键值对RDD中包含的是一个个二元组,所以需要传递的函数会由原来的操作单个元素改为操作二元组。

下表总结了针对单个键值对RDD的转化操作,以 { (1,2) , (3,4) , (3,6) }  为例,f表示传入的函数

函数名 目的 示例 结果
reduceByKey(f) 合并具有相同key的值 rdd.reduceByKey( ( x,y) => x+y ) { (1,2) , (3,10) }
groupByKey() 对具有相同key的值分组 rdd.groupByKey() { (1,2) , (3, [4,6] ) }
mapValues(f) 对键值对中的每个值(value)应用一个函数,但不改变键(key) rdd.mapValues(x => x+1) { (1,3) , (3,5) , (3,7) }
combineBy Key( createCombiner, mergeValue, mergeCombiners, partitioner) 使用不同的返回类型合并具有相同键的值 下面有详细讲解 -
flatMapValues(f) 对键值对RDD中每个值应用返回一个迭代器的函数,然后对每个元素生成一个对应的键值对。常用语符号化 rdd.flatMapValues(x => ( x to 5 ))

{ (1, 2) ,  (1, 3) ,   (1, 4) , (1, 5) ,  (3, 4) , (3, 5) }

keys() 获取所有key rdd.keys() {1,3,3}
values() 获取所有value rdd.values() {2,4,6}
sortByKey() 根据key排序 rdd.sortByKey() { (1,2) , (3,4) , (3,6) }

 

 

下表总结了针对两个键值对RDD的转化操作,以rdd1 = { (1,2) , (3,4) , (3,6) }  rdd2 = { (3,9) } 为例,

函数名 目的 示例 结果
subtractByKey 删掉rdd1中与rdd2的key相同的元素 rdd1.subtractByKey(rdd2) { (1,2) }
join 内连接 rdd1.join(rdd2)

{(3, (4, 9)), (3, (6, 9))}

leftOuterJoin 左外链接 rdd1.leftOuterJoin (rdd2)

{(3,( Some( 4), 9)), (3,( Some( 6), 9))}

rightOuterJoin 右外链接 rdd1.rightOuterJoin(rdd2)

{(1,( 2, None)), (3, (4, Some( 9))), (3, (6, Some( 9)))}

cogroup 将两个RDD钟相同key的数据分组到一起 rdd1.cogroup(rdd2) {(1,([ 2],[])), (3, ([4, 6],[ 9]))}

 

 

Setp2

combineByKey

combineByKey( createCombiner, mergeValue, mergeCombiners, partitioner,mapSideCombine)

combineByKey( createCombiner, mergeValue, mergeCombiners, partitioner)

combineByKey( createCombiner, mergeValue, mergeCombiners)

 

函数功能:

聚合各分区的元素,而每个元素都是二元组。功能与基础RDD函数aggregate()差不多,可让用户返回与输入数据类型不同的返回值。

combineByKey函数的每个参数分别对应聚合操作的各个阶段。所以,理解此函数对Spark如何操作RDD会有很大帮助。

 

参数解析:

createCombiner:分区内 创建组合函数

mergeValue:分区内 合并值函数

mergeCombiners:多分区 合并组合器函数

partitioner:自定义分区数,默认为HashPartitioner

mapSideCombine:是否在map端进行Combine操作,默认为true

 

工作流程:

  1. combineByKey会遍历分区中的所有元素,因此每个元素的key要么没遇到过,要么和之前某个元素的key相同。
  2. 如果这是一个新的元素,函数会调用createCombiner创建那个key对应的累加器初始值

  3. 如果这是一个在处理当前分区之前已经遇到的key,会调用mergeCombiners把该key累加器对应的当前value与这个新的value合并

 

代码例子:

//统计男女个数

 val conf = new SparkConf ().setMaster ("local").setAppName ("app_1")
    val sc = new SparkContext (conf)

    val people = List(("男", "李四"), ("男", "张三"), ("女", "韩梅梅"), ("女", "李思思"), ("男", "马云"))
    val rdd = sc.parallelize(people,2)
    val result = rdd.combineByKey(
      (x: String) => (List(x), 1),  //createCombiner
      (peo: (List[String], Int), x : String) => (x :: peo._1, peo._2 + 1), //mergeValue
      (sex1: (List[String], Int), sex2: (List[String], Int)) => (sex1._1 ::: sex2._1, sex1._2 + sex2._2)) //mergeCombiners
    result.foreach(println)

结果

(男, ( List( 张三,  李四,  马云),3 ) )
(女, ( List( 李思思,  韩梅梅),2 ) )

 

流程分解:

图片 1

 

解析:两个分区,分区一按顺序V1、V2、V3遍历

  • V1,发现第一个key=男时,调用createCombiner,即

    (x: String) => (List(x), 1)
    
  • V2,第二次碰到key=男的元素,调用mergeValue,即

    (peo: (List[String], Int), x : String) => (x :: peo._1, peo._2 + 1)
    
  • V3,发现第一个key=女,继续调用createCombiner,即

    (x: String) => (List(x), 1)
    
  • … …

  • 待各V1、V2分区都计算完后,数据进行混洗,调用mergeCombiners,即

    (sex1: (List[String], Int), sex2: (List[String], Int)) => (sex1._1 ::: sex2._1, sex1._2 + sex2._2))
    

 


 

add by jan 2017-02-27 18:34:39

以下例子都基于此RDD

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

reduceByKey(func)

reduceByKey(func)的功能是,使用func函数合并具有相同键的值。

比如,reduceByKey((a,b) => a+b),有四个键值对("spark",1)、("spark",2)、("hadoop",3)和("hadoop",5),对具有相同key的键值对进行合并后的结果就是:("spark",3)、("hadoop",8)。可以看出,(a,b) => a+b这个Lamda表达式中,a和b都是指value,比如,对于两个具有相同key的键值对("spark",1)、("spark",2),a就是1,b就是2。

scala> pairRDD.reduceByKey((a,b)=>a+b).foreach(println)
(Spark,2)
(Hive,1)
(Hadoop,1)

  

Setp3

groupByKey()

roupByKey()的功能是,对具有相同键的值进行分组。比如,对四个键值对("spark",1)、("spark",2)、("hadoop",3)和("hadoop",5),采用groupByKey()后得到的结果是:("spark",(1,2))和("hadoop",(3,5))。

scala> pairRDD.groupByKey()
res15: org.apache.spark.rdd.RDD[(String, Iterable[Int])] = ShuffledRDD[15] at groupByKey at <console>:34
//从上面执行结果信息中可以看出,分组后,value被保存到Iterable[Int]中
scala> pairRDD.groupByKey().foreach(println)
(Spark,CompactBuffer(1, 1))
(Hive,CompactBuffer(1))
(Hadoop,CompactBuffer(1))

  

def combineByKeyWithClassTag[C]( createCombiner: V => C, mergeValue:  => C, mergeCombiners:  => C, partitioner: Partitioner, mapSideCombine: Boolean = true, serializer: Serializer = null)(implicit ct: ClassTag[C]): RDD[] = self.withScope { require(mergeCombiners != null, "mergeCombiners must be defined") // required as of Spark 0.9.0 if (keyClass.isArray) { if (mapSideCombine) { throw new SparkException("Cannot use map-side combining with array keys.") } if (partitioner.isInstanceOf[HashPartitioner]) { throw new SparkException("HashPartitioner cannot partition array keys.") } } val aggregator = new Aggregator[K, V, C]( self.context.clean(createCombiner), self.context.clean(mergeValue), self.context.clean(mergeCombiners)) if (self.partitioner == Some(partitioner)) { self.mapPartitions(iter => { val context = TaskContext.get() new InterruptibleIterator(context, aggregator.combineValuesByKey(iter, context)) }, preservesPartitioning = true) } else { new ShuffledRDD[K, V, C](self, partitioner) .setSerializer(serializer) .setAggregator(aggregator) .setMapSideCombine(mapSideCombine) } }

keys

keys只会把键值对RDD中的key返回形成一个新的RDD。比如,对四个键值对("spark",1)、("spark",2)、("hadoop",3)和("hadoop",5)构成的RDD,采用keys后得到的结果是一个RDD[Int],内容是{"spark","spark","hadoop","hadoop"}。

scala> pairRDD.keys
res17: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[17] at keys at <console>:34
scala> pairRDD.keys.foreach(println)
Hadoop
Spark
Hive
Spark

  

姑且不去看方法里面的细节,我们会只要知道最后调用的是 combineByKeyWithClassTag 这个方法。这个方法有两个参数我们来重点看一下,

values

 values只会把键值对RDD中的value返回形成一个新的RDD。比如,对四个键值对("spark",1)、("spark",2)、("hadoop",3)和("hadoop",5)构成的RDD,采用keys后得到的结果是一个RDD[Int],内容是{1,2,3,5}。

scala> pairRDD.values
res0: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[2] at values at <console>:34

scala> pairRDD.values.foreach(println)
1
1
1
1

  

def combineByKeyWithClassTag[C]( createCombiner: V => C, mergeValue:  => C, mergeCombiners:  => C, partitioner: Partitioner, mapSideCombine: Boolean = true, serializer: Serializer = null)

sortByKey()

 sortByKey()的功能是返回一个根据键排序的RDD。

scala> pairRDD.sortByKey()
res0: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[2] at sortByKey at <console>:34
scala> pairRDD.sortByKey().foreach(println)
(Hadoop,1)
(Hive,1)
(Spark,1)
(Spark,1)

  

首先是 partitioner 参数 ,这个即是 RDD 的分区设置。除了默认的 defaultPartitioner,Spark 还提供了 RangePartitioner 和 HashPartitioner 外,此外用户也可以自定义 partitioner 。通过源码可以发现如果是 HashPartitioner 的话,那么是会抛出一个错误的。

mapValues(func)

我们经常会遇到一种情形,我们只想对键值对RDD的value部分进行处理,而不是同时对key和value进行处理。对于这种情形,Spark提供了mapValues(func),它的功能是,对键值对RDD中的每个value都应用一个函数,但是,key不会发生变化。比如,对四个键值对("spark",1)、("spark",2)、("hadoop",3)和("hadoop",5)构成的pairRDD,如果执行pairRDD.mapValues(x => x+1),就会得到一个新的键值对RDD,它包含下面四个键值对("spark",2)、("spark",3)、("hadoop",4)和("hadoop",6)。 

scala> pairRDD.mapValues(x => x+1)
res2: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[4] at mapValues at <console>:34
scala> pairRDD.mapValues(x => x+1).foreach(println)
(Hadoop,2)
(Spark,2)
(Hive,2)
(Spark,2)

  

然后是 mapSideCombine 参数 ,这个参数正是 reduceByKey 和 groupByKey 最大不同的地方,它决定是是否会先在节点上进行一次 Combine 操作,下面会有更具体的例子来介绍。

join

join(连接)操作是键值对常用的操作。“连接”(join)这个概念来自于关系数据库领域,因此,join的类型也和关系数据库中的join一样,包括内连接(join)、左外连接(leftOuterJoin)、右外连接(rightOuterJoin)等。最常用的情形是内连接,所以,join就表示内连接。
对于内连接,对于给定的两个输入数据集(K,V1)和(K,V2),只有在两个数据集中都存在的key才会被输出,最终得到一个(K,(V1,V2))类型的数据集。

比如,pairRDD1是一个键值对集合{("spark",1)、("spark",2)、("hadoop",3)和("hadoop",5)},pairRDD2是一个键值对集合{("spark","fast")},那么,pairRDD1.join(pairRDD2)的结果就是一个新的RDD,这个新的RDD是键值对集合{("spark",1,"fast"),("spark",2,"fast")}。对于这个实例,我们下面在spark-shell中运行一下:

scala> val pairRDD1 = sc.parallelize(Array(("spark",1),("spark",2),("hadoop",3),("hadoop",5)))
pairRDD1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[24] at parallelize at <console>:27

scala> val pairRDD2 = sc.parallelize(Array(("spark","fast")))
pairRDD2: org.apache.spark.rdd.RDD[(String, String)] = ParallelCollectionRDD[25] at parallelize at <console>:27

scala> pairRDD1.join(pairRDD2)
res9: org.apache.spark.rdd.RDD[(String, (Int, String))] = MapPartitionsRDD[28] at join at <console>:32

scala> pairRDD1.join(pairRDD2).foreach(println)
(spark,(1,fast))
(spark,(2,fast))

  

 

 

 

详细请参考《Spark快速大数据分析》

Step1

 def groupByKey(): RDD[(K, Iterable[V])] = self.withScope { groupByKey(defaultPartitioner }

Step2

 def groupByKey(partitioner: Partitioner): RDD[(K, Iterable[V])] = self.withScope { // groupByKey shouldn't use map side combine because map side combine does not // reduce the amount of data shuffled and requires all map side data be inserted // into a hash table, leading to more objects in the old gen. val createCombiner =  => CompactBuffer 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])]] }

Setp3

def combineByKeyWithClassTag[C]( createCombiner: V => C, mergeValue:  => C, mergeCombiners:  => C, partitioner: Partitioner, mapSideCombine: Boolean = true, serializer: Serializer = null)(implicit ct: ClassTag[C]): RDD[] = self.withScope { require(mergeCombiners != null, "mergeCombiners must be defined") // required as of Spark 0.9.0 if (keyClass.isArray) { if (mapSideCombine) { throw new SparkException("Cannot use map-side combining with array keys.") } if (partitioner.isInstanceOf[HashPartitioner]) { throw new SparkException("HashPartitioner cannot partition array keys.") } } val aggregator = new Aggregator[K, V, C]( self.context.clean(createCombiner), self.context.clean(mergeValue), self.context.clean(mergeCombiners)) if (self.partitioner == Some(partitioner)) { self.mapPartitions(iter => { val context = TaskContext.get() new InterruptibleIterator(context, aggregator.combineValuesByKey(iter, context)) }, preservesPartitioning = true) } else { new ShuffledRDD[K, V, C](self, partitioner) .setSerializer(serializer) .setAggregator(aggregator) .setMapSideCombine(mapSideCombine) } }

结合上面 reduceByKey 的调用链,可以发现最终其实都是调用 combineByKeyWithClassTag 这个方法的,但调用的参数不同。reduceByKey的调用

combineByKeyWithClassTag[V] => v, func, func, partitioner)

groupByKey的调用

combineByKeyWithClassTag[CompactBuffer[V]]( createCombiner, mergeValue, mergeCombiners, partitioner, mapSideCombine = false)

正是两者不同的调用方式导致了两个方法的差别,我们分别来看

  • reduceByKey的泛型参数直接是[V],而groupByKey的泛型参数是[CompactBuffer[V]]。这直接导致了 reduceByKey 和 groupByKey 的返回值不同,前者是RDD[],而后者是RDD[(K, Iterable[V])]

  • 然后就是mapSideCombine = false 了,这个mapSideCombine 参数的默认是true的。这个值有什么用呢,上面也说了,这个参数的作用是控制要不要在map端进行初步合并。可以看看下面具体的例子。

<img src="" width="65%" />

<img src="" width="65%" />

从功能上来说,可以发现 ReduceByKey 其实就是会在每个节点先进行一次合并的操作,而 groupByKey 没有。

这么来看 ReduceByKey 的性能会比 groupByKey 好很多,因为有些工作在节点已经处理了。那么 groupByKey 为什么存在,它的应用场景是什么呢?我也不清楚,如果观看这篇文章的读者知道的话不妨在评论里说出来吧。非常感谢!

本文由时时彩平台发布于编程,转载请注明出处:spark RDD,reduceByKey vs groupByKey

关键词:

上一篇:Spring Boot 2.0 教程系列 | 配置 Undertow 容器

下一篇:没有了