文档章节

spark RDD,reduceByKey vs groupByKey

终日而思一
 终日而思一
发布于 2018/11/20 17:48
字数 1063
阅读 22
收藏 0

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

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

所用 spark 版本:spark 2.1.0

#### 先看reduceByKey
Step1
```
  def reduceByKey(func: (V, V) => V): RDD[(K, V)] = self.withScope {
    reduceByKey(defaultPartitioner(self), func)
  }
```
Setp2
```
  def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)] = self.withScope {
    combineByKeyWithClassTag[V]((v: V) => v, func, func, partitioner)
  }
```
Setp3
```
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 {
    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)
    }
  }
```

姑且不去看方法里面的细节,我们会只要知道最后调用的是 combineByKeyWithClassTag 这个方法。这个方法有两个参数我们来重点看一下,
```
def combineByKeyWithClassTag[C](
      createCombiner: V => C,
      mergeValue: (C, V) => C,
      mergeCombiners: (C, C) => C,
      partitioner: Partitioner,
      mapSideCombine: Boolean = true,
      serializer: Serializer = null)
```
首先是 **partitioner** 参数 ,这个即是 RDD 的分区设置。除了默认的 defaultPartitioner,Spark 还提供了 RangePartitioner 和 HashPartitioner 外,此外用户也可以自定义 partitioner 。通过源码可以发现如果是 HashPartitioner 的话,那么是会抛出一个错误的。

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

#### 然后是groupByKey
Step1
```
  def groupByKey(): RDD[(K, Iterable[V])] = self.withScope {
    groupByKey(defaultPartitioner(self))
  }
```
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 = (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])]]
  }
```
Setp3
```
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 {
    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: V) => v, func, func, partitioner)
```
groupByKey的调用
```
combineByKeyWithClassTag[CompactBuffer[V]](
      createCombiner, mergeValue, mergeCombiners, partitioner, mapSideCombine = false)
```
正是两者不同的调用方式导致了两个方法的差别,我们分别来看
- reduceByKey的泛型参数直接是[V],而groupByKey的泛型参数是[CompactBuffer[V]]。这直接导致了 reduceByKey 和 groupByKey 的返回值不同,前者是RDD[(K, V)],而后者是RDD[(K, Iterable[V])]

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


<img src="https://img2018.cnblogs.com/blog/1011838/201810/1011838-20181027094209726-2067397752.png" width="65%" />

<img src="https://img2018.cnblogs.com/blog/1011838/201810/1011838-20181027094214131-2054597375.png" width="65%" />

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

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

© 著作权归作者所有

终日而思一
粉丝 2
博文 20
码字总数 33151
作品 0
广州
私信 提问
spark RDD,reduceByKey vs groupByKey

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

终日而思一
2018/10/28
0
0
【Spark】弹性分布式数据集RDD及其操作

版权声明:本文为博主原创文章,转载请注明出处。 https://blog.csdn.net/gongxifacai_believe/article/details/86714557 1、RDD简介 RDD是Spark提供的核心抽象,全称为Resillient Distribu...

魏晓蕾
01/31
0
0
影响Spark输出RDD分区的操作函数

会影响到Spark输出RDD分区(partitioner)的操作 cogroup, groupWith, join, leftOuterJoin, rightOuterJoin, groupByKey, reduceByKey, combineByKey, partitionBy, sort, mapValues(如果父......

imbetter
2018/04/22
0
0
Spark中的键值对操作 JavaPairRDD

1.PairRDD介绍 List list=new ArrayList(); list.add("this is a test"); list.add("how are you?"); list.add("do you love me?"); list.add("can you tell me?"); JavaRDD lines=sc.paral......

u014236541
2018/05/08
0
0
Spark技术内幕:究竟什么是RDD

版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/anzhsoft2008/article/details/39851421 RDD是Spark最基本,也是最根本的数据抽象。http://www.cs.berkeley....

anzhsoft
2014/10/07
0
0

没有更多内容

加载失败,请刷新页面

加载更多

Replugin借助“UI进程”来快速释放Dex

public static boolean preload(PluginInfo pi) { if (pi == null) { return false; } // 借助“UI进程”来快速释放Dex(见PluginFastInstallProviderProxy的说明) return PluginFastInsta......

Gemini-Lin
46分钟前
4
0
Hibernate 5 的模块/包(modules/artifacts)

Hibernate 的功能被拆分成一系列的模块/包(modules/artifacts),其目的是为了对依赖进行独立(模块化)。 模块名称 说明 hibernate-core 这个是 Hibernate 的主要(main (core))模块。定义...

honeymoose
今天
4
0
CSS--属性

一、溢出 当内容多,元素区域小的时候,就会产生溢出效果,默认是纵向溢出 横向溢出:在内容和容器之间再套一层容器,并且内部容器要比外部容器宽 属性:overflow/overflow-x/overflow-y 取值...

wytao1995
今天
4
0
精华帖

第一章 jQuery简介 jQuery是一个JavaScript库 jQuery具备简洁的语法和跨平台的兼容性 简化了JavaScript的操作。 在页面中引入jQuery jQuery是一个JavaScript脚本库,不需要特别的安装,只需要...

流川偑
今天
7
0
语音对话英语翻译在线翻译成中文哪个方法好用

想要进行将中文翻译成英文,或者将英文翻译成中文的操作,其实有一个非常简单的工具就能够帮助完成将语音进行翻译转换的软件。 在应用市场或者百度手机助手等各大应用渠道里面就能够找到一款...

401恶户
今天
3
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部