如何使用spark sql过滤特定聚合的行? [英] How to filter rows for a specific aggregate with spark sql?

问题描述

通常一个组中的所有行都传递给一个聚合函数.我想使用条件过滤行，以便仅将组中的某些行传递给聚合函数.使用 PostgreSQL 可以进行此类操作.我想用 Spark SQL DataFrame (Spark 2.0.0) 做同样的事情.

Normally all rows in a group are passed to an aggregate function. I would like to filter rows using a condition so that only some rows within a group are passed to an aggregate function. Such operation is possible with PostgreSQL. I would like to do the same thing with Spark SQL DataFrame (Spark 2.0.0).

代码可能如下所示:

val df = ... // some data frame
df.groupBy("A").agg(
  max("B").where("B").less(10), // there is no such method as `where` :(
  max("C").where("C").less(5)
)

对于这样的数据框:

| A | B | C |
|  1| 14|  4|
|  1|  9|  3|
|  2|  5|  6|

结果是:

|A|max(B)|max(C)|
|1|    9|      4|
|2|    5|   null|

可以使用 Spark SQL 吗?

Is it possible with Spark SQL?

请注意，通常可以使用除 max 之外的任何其他聚合函数，并且可以在具有任意过滤条件的同一列上存在多个聚合.

Note that in general any other aggregate function than max could be used and there could be multiple aggregates over the same column with arbitrary filtering conditions.

推荐答案

val df = Seq(
    (1,14,4),
    (1,9,3),
    (2,5,6)
  ).toDF("a","b","c")

val aggregatedDF = df.groupBy("a")
  .agg(
    max(when($"b" < 10, $"b")).as("MaxB"),
    max(when($"c" < 5, $"c")).as("MaxC")
  )

aggregatedDF.show

这篇关于如何使用spark sql过滤特定聚合的行?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！