【Spark系列2】reduceByKey和groupByKey区别与用法

2023-11-17

在spark中,我们知道一切的操作都是基于RDD的。在使用中,RDD有一种非常特殊也是非常实用的format——pair RDD,即RDD的每一行是(key, value)的格式。这种格式很像Python的字典类型,便于针对key进行一些处理。


针对pair RDD这样的特殊形式,spark中定义了许多方便的操作,今天主要介绍一下reduceByKey和groupByKey,因为在接下来讲解《在spark中如何实现SQL中的group_concat功能?》时会用到这两个operations。


首先,看一看spark官网[1]是怎么解释的:

reduceByKey(func, numPartitions=None)

Merge the values for each key using an associative reduce function. This will also perform the merginglocally on each mapper before sending results to a reducer, similarly to a “combiner” in MapReduce. Output will be hash-partitioned with numPartitions partitions, or the default parallelism level if numPartitions is not specified.

也就是,reduceByKey用于对每个key对应的多个value进行merge操作,最重要的是它能够在本地先进行merge操作,并且merge操作可以通过函数自定义。

groupByKey(numPartitions=None)

Group the values for each key in the RDD into a single sequence. Hash-partitions the resulting RDD with numPartitions partitions. Note: If you are grouping in order to perform an aggregation (such as a sum or average) over each key, using reduceByKey or aggregateByKey will provide much better performance.

也就是,groupByKey也是对每个key进行操作,但只生成一个sequence。需要特别注意“Note”中的话,它告诉我们:如果需要对sequence进行aggregation操作(注意,groupByKey本身不能自定义操作函数),那么,选择reduceByKey/aggregateByKey更好。这是因为groupByKey不能自定义函数,我们需要先用groupByKey生成RDD,然后才能对此RDD通过map进行自定义函数操作。


为了更好的理解上面这段话,下面我们使用两种不同的方式去计算单词的个数[2]:

val words = Array("one", "two", "two", "three", "three", "three")

val wordPairsRDD = sc.parallelize(words).map(word => (word, 1))

val wordCountsWithReduce = wordPairsRDD.reduceByKey(_ + _)

val wordCountsWithGroup = wordPairsRDD.groupByKey().map(t => (t._1, t._2.sum))
上面得到的wordCountsWithReduce和wordCountsWithGroup是完全一样的,但是,它们的内部运算过程是不同的。


(1)当采用reduceByKeyt时,Spark可以在每个分区移动数据之前将待输出数据与一个共用的key结合。借助下图可以理解在reduceByKey里究竟发生了什么。 注意在数据对被搬移前同一机器上同样的key是怎样被组合的(reduceByKey中的lamdba函数)。然后lamdba函数在每个区上被再次调用来将所有值reduce成一个最终结果。整个过程如下:



(2)当采用groupByKey时,由于它不接收函数,spark只能先将所有的键值对(key-value pair)都移动,这样的后果是集群节点之间的开销很大,导致传输延时。整个过程如下:


因此,在对大数据进行复杂计算时,reduceByKey优于groupByKey

另外,如果仅仅是group处理,那么以下函数应该优先于 groupByKey
  (1)、combineByKey 组合数据,但是组合之后的数据类型与输入时值的类型不一样。
  (2)、foldByKey合并每一个 key 的所有值,在级联函数和“零值”中使用。


最后,对reduceByKey中的func做一些介绍:

如果是用Python写的spark,那么有一个库非常实用:operator[3],其中可以用的函数包括:大小比较函数,逻辑操作函数,数学运算函数,序列操作函数等等。这些函数可以直接通过“from operator import *”进行调用,直接把函数名作为参数传递给reduceByKey即可。如下:

<span style="font-size:14px;">from operator import add
rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 1)])
sorted(rdd.reduceByKey(add).collect())

[('a', 2), ('b', 1)]</span>



参考:

[1] http://spark.apache.org/docs/latest/api/python/pyspark.html?highlight=reducebykey#pyspark.RDD.reduceByKey

[2] http://www.iteblog.com/archives/1357

[3] https://docs.python.org/2/library/operator.html

本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)

【Spark系列2】reduceByKey和groupByKey区别与用法 的相关文章

  • spark集群搭建与mysql元数据管理

    找个spark集群搭建是针对于上一篇hadoop的基础上搭建的 所以spark的版本也是要按照着hadoop版本进行下载 1 解压spark 修改spark的 etc profile的home目录 2 安装SCALA 并配置SCALA HO
  • Spark(七)——累加器和广播变量

    5 累加器 通过在驱动器中调用SparkContext accumulator initialValue 方法 创建出存有初始值的累加器 返回值为org apache spark Accumulator T 对象 其中 T 是初始值 ini
  • Spark基础知识(个人总结)

    声明 1 本文为我的个人复习总结 并非那种从零基础开始普及知识 内容详细全面 言辞官方的文章 2 由于是个人总结 所以用最精简的话语来写文章 3 若有错误不当之处 请指出 一 Spark概述 Spark模块 Core SQL Streami
  • 11.Linux下Spark的安装配置以及spark-shell的启动和 Spark集群环境搭建

    本案例软件包 链接 https pan baidu com s 1zABhjj2umontXe2CYBW DQ 提取码 1123 若链接失效在下面评论 我会及时更新 目录 1 安装Spark 1 先用xftp将安装包传到home hadoo
  • Spark数据处理实战之列空值&新增列处理

    在spark dataset dataframe操作过程中 我们经常会遇到对于一个列值的一些判断情况 是否为NULL 创建一个新列等 本文讲解了常用的增加列的方法 并且对于列空值判断 填充处理以及查询的api做了详细的描述和实例讲解 阅读完
  • 学习大数据spark——心得体会

    总结与体会 1 项目总结 本次项目实现了Spark 单机模式Python版的安装 介绍了与Spark编程有关的一些基本概念 特别对RDD的创建 转换和行动操作做了比较详细的说明 对从RDD 到DataFrame的实现进 行了案例训练 包括
  • dolphinschedule使用shell任务结束状态研究

    背景 配置的dolphin任务 使用的是shell shell里包含了spark submit 如下截图 dolphin shell 介绍完毕 开始说明现象 有天有人调整了集群的cdp配置 executor cores max 1 我之前这
  • 【Apache Spark 】第 1 章Apache Spark 简介:统一分析引擎

    大家好 我是Sonhhxg 柒 希望你看完之后 能对你有所帮助 不足请指正 共同学习交流 个人主页 Sonhhxg 柒的博客 CSDN博客 欢迎各位 点赞 收藏 留言 系列专栏 机器学习 ML 自然语言处理 NLP 深度学习 DL fore
  • spark算子执行位置研究,driver端?executor端?

    参考资料 https cloud tencent com developer article 1545723 前言 spark算子的执行位置 driver端 还是executor端 这些之前其实没有注意过 最近在学流处理 发现这个还是很重要
  • Spark SQL 之 Temporary View

    Spark SQL 之 Temporary View spark SQL的 temporary view 是支持原生SQL 的方式之一 spark SQL的 DataFrame 和 DataSet 均可以通过注册 temporary vie
  • 【Spark NLP】第 7 章:分类和回归

    大家好 我是Sonhhxg 柒 希望你看完之后 能对你有所帮助 不足请指正 共同学习交流 个人主页 Sonhhxg 柒的博客 CSDN博客 欢迎各位 点赞 收藏 留言 系列专栏 机器学习 ML 自然语言处理 NLP 深度学习 DL fore
  • 大数据开发必备面试题Spark篇合集

    1 Hadoop 和 Spark 的相同点和不同点 Hadoop 底层使用 MapReduce 计算架构 只有 map 和 reduce 两种操作 表达能力比较欠缺 而且在 MR 过程中会重复的读写 hdfs 造成大量的磁盘 io 读写操作
  • 大数据—— Flink 的优化

    目录 一 Flink内存优化 1 1 Flink 内存配置 二 配置进程参数 2 1 场景 2 2 操作步骤 三 解决数据倾斜 3 1 场景描述 3 2 解决方式 3 2 1 数据源的消费不均匀 调整并发度 3 2 2 数据分布不均匀 四
  • 使用Flink1.16.0的SQLGateway迁移Hive SQL任务

    使用Flink的SQL Gateway迁移Hive SQL任务 前言 我们有数万个离线任务 主要还是默认的DataPhin调度CDP集群的Hive On Tez这种低成本任务 当然也有PySpark 打Jar包的Spark和打Jar包的Fl
  • Flume之:二、企业开发案例

    Flume之 二 企业开发案例 文章目录 Flume之 二 企业开发案例 三 企业开发案例 1 监控端口数据官方案例 2 实时读取本地文件到HDFS案例 3 实时读取目录文件到HDFS案例 4 flume监控Kafka gt Spark知识
  • 2020-10-24 大数据面试问题

    上周面试数据开发职位主要从公司的视角讲一下记录下面试流水 1 三面技术一轮hr 面到了cto 整体来看是这一周技术含量最高信息量最大的一个 1到4轮过了4个小时 技术上的问题主要问的对数据分层的理解 1 一面自我介绍 目前团队的规模多大 2
  • sparkstreamming 消费kafka(2)

    spark streaming提供了两种获取方式 一种是同storm一样 实时读取缓存到内存中 另一种是定时批量读取 这两种方式分别是 Receiver base Direct 一 Receiver base Spark官方最先提供了基于R
  • Spark Sql之dropDuplicates去重

    文章目录 算子介绍 示例 问题 解决 dropDuplicates和distinct 参考 算子介绍 dropDuplicates去重原则 按数据行的顺序保留每行数据出现的第一条 dropDuplicates 在Spark源码里面提供了以下
  • JAVA 安装与简单使用

    JAVA简易安装 下载安装 环境变量 进入变量界面 设置变量 验证JAVA环境 运行Java程序 个人站 ghzzz cn 还在备案 很快就能访问了 下载安装 第一步当然是从官网下载安装java了 网上有很多的教程 这里简单的写一下 在这里
  • 2023_Spark_实验二十九:Flume配置KafkaSink

    实验目的 掌握Flume采集数据发送到Kafka的方法 实验方法 通过配置Flume的KafkaSink采集数据到Kafka中 实验步骤 一 明确日志采集方式 一般Flume采集日志source有两种方式 1 Exec类型的Source 可

随机推荐