Spark DataFrame 的 groupBy vs groupByKey

本文涉及的产品
EMR Serverless StarRocks,5000CU*H 48000GB*H
简介:


在使用 Spark SQL 的过程中,经常会用到 groupBy 这个函数进行一些统计工作。但是会发现除了 groupBy 外,还有一个 groupByKey(注意RDD 也有一个 groupByKey,而这里的 groupByKey 是 DataFrame 的 ) 。这个 groupByKey 引起了我的好奇,那我们就到源码里面一探究竟吧。

所用 spark 版本:spark 2.1.0

先从使用的角度来说,
groupBy:groupBy类似于传统SQL语言中的group by子语句,但比较不同的是groupBy()可以带多个列名,对多个列进行group。比如想根据 "id" 和 "name" 进行 groupBy 的话可以

df.goupBy("id","name")

groupBy返回的类型是RelationalGroupedDataset。

groupByKey:groupByKey则更加灵活,可以根据用户自己对列的组合来进行groupBy,比如上面的那个例子,根据 "id" 和 "name" 进行 groupBy,使用groupByKey可以这样。

//同前面的goupBy效果是一样的,但返回的类型是不一样的
df..toDF("id","name").goupByKey(row =>{
    row.getString(0) + row.getString(1)
})

但和groupBy不同的是groupByKey返回的类型是KeyValueGroupedDataset。

下面来看看这两个方法的实现有何区别。

groupBy
  def groupBy(cols: Column*): RelationalGroupedDataset = {
    RelationalGroupedDataset(toDF(), cols.map(_.expr), RelationalGroupedDataset.GroupByType)
  }

最终会去新建一个RelationalGroupedDataset,而这个方法提供count(),max(),agg(),等方法。值得一提的是,这个类在spark1.x的时候类名为“GroupedData”。看看类中的注释吧

/**
 * A set of methods for aggregations on a `DataFrame`, created by `Dataset.groupBy`.
 *
 * The main method is the agg function, which has multiple variants. This class also contains
 * convenience some first order statistics such as mean, sum for convenience.
 *
 * This class was named `GroupedData` in Spark 1.x.
 *
 * @since 2.0.0
 */
@InterfaceStability.Stable
class RelationalGroupedDataset protected[sql](
groupByKey
  @Experimental
  @InterfaceStability.Evolving
  def groupByKey[K: Encoder](func: T => K): KeyValueGroupedDataset[K, T] = {
    val inputPlan = logicalPlan
    val withGroupingKey = AppendColumns(func, inputPlan)
    val executed = sparkSession.sessionState.executePlan(withGroupingKey)

    new KeyValueGroupedDataset(
      encoderFor[K],
      encoderFor[T],
      executed,
      inputPlan.output,
      withGroupingKey.newColumns)
  }

可以发现最后生成和返回的类是KeyValueGroupedDataset。这是dataset的子类,表示聚合过之后的dataset。
我们再看看这个类中的注释吧

/**
 * :: Experimental ::
 * A [[Dataset]] has been logically grouped by a user specified grouping key.  Users should not
 * construct a [[KeyValueGroupedDataset]] directly, but should instead call `groupByKey` on
 * an existing [[Dataset]].
 *
 * @since 2.0.0
 */
@Experimental
@InterfaceStability.Evolving
class KeyValueGroupedDataset[K, V] private[sql](

可以发现 groupByKey 还处于实验阶段。它是希望可以由用户自己来实现 groupBy 的规则,而不像 groupBy() 一样,需要被列属性所束缚。
通过 groupByKey 用户可以按照自己的需求来进行 grouping 。

总而言之,groupByKey虽然提供了更加灵活的处理 grouping 的方式,但 groupByKey 后返回的类是 KeyValueGroupedDataset ,它里面所提供的操作接口也不如 groupBy 返回的 RelationalGroupedDataset 所提供的接口丰富。除非真的有一些特殊的 grouping 操作,否则还是使用 groupBy 吧。

相关实践学习
基于EMR Serverless StarRocks一键玩转世界杯
基于StarRocks构建极速统一OLAP平台
快速掌握阿里云 E-MapReduce
E-MapReduce 是构建于阿里云 ECS 弹性虚拟机之上,利用开源大数据生态系统,包括 Hadoop、Spark、HBase,为用户提供集群、作业、数据等管理的一站式大数据处理分析服务。 本课程主要介绍阿里云 E-MapReduce 的使用方法。
相关文章
|
6月前
|
SQL 分布式计算 数据库
【大数据技术Spark】Spark SQL操作Dataframe、读写MySQL、Hive数据库实战(附源码)
【大数据技术Spark】Spark SQL操作Dataframe、读写MySQL、Hive数据库实战(附源码)
250 0
|
6月前
|
SQL 分布式计算 大数据
【大数据技术Hadoop+Spark】Spark SQL、DataFrame、Dataset的讲解及操作演示(图文解释)
【大数据技术Hadoop+Spark】Spark SQL、DataFrame、Dataset的讲解及操作演示(图文解释)
169 0
|
9天前
|
SQL JSON 分布式计算
【赵渝强老师】Spark SQL的数据模型:DataFrame
本文介绍了在Spark SQL中创建DataFrame的三种方法。首先,通过定义case class来创建表结构,然后将CSV文件读入RDD并关联Schema生成DataFrame。其次,使用StructType定义表结构,同样将CSV文件读入RDD并转换为Row对象后创建DataFrame。最后,直接加载带有格式的数据文件(如JSON),通过读取文件内容直接创建DataFrame。每种方法都包含详细的代码示例和解释。
|
1月前
|
SQL 分布式计算 大数据
大数据-94 Spark 集群 SQL DataFrame & DataSet & RDD 创建与相互转换 SparkSQL
大数据-94 Spark 集群 SQL DataFrame & DataSet & RDD 创建与相互转换 SparkSQL
53 0
|
3月前
|
SQL 存储 分布式计算
|
6月前
|
SQL 分布式计算 数据挖掘
Spark_Day07:Spark SQL(DataFrame是什么和数据分析(案例讲解))
Spark_Day07:Spark SQL(DataFrame是什么和数据分析(案例讲解))
173 0
|
6月前
|
JSON 分布式计算 关系型数据库
Spark中使用DataFrame进行数据转换和操作
Spark中使用DataFrame进行数据转换和操作
|
6月前
|
SQL 分布式计算 安全
Spark的核心概念:RDD、DataFrame和Dataset
Spark的核心概念:RDD、DataFrame和Dataset
|
6月前
|
SQL 分布式计算 Java
Spark中的DataFrame和Dataset有什么区别?请解释其概念和用途。
Spark中的DataFrame和Dataset有什么区别?请解释其概念和用途。
205 0
|
6月前
|
SQL 分布式计算 关系型数据库
Spark【Spark SQL(二)RDD转换DataFrame、Spark SQL读写数据库 】
Spark【Spark SQL(二)RDD转换DataFrame、Spark SQL读写数据库 】