Spark DataFrame 的 groupBy vs groupByKey-白红宇

在使用 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.Stableclass 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.Evolvingclass KeyValueGroupedDataset[K, V] private[sql](

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

通过 groupByKey 用户可以按照自己的需求来进行 grouping 。

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