如何使用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).

代码可能看起来像这样：

The code could probably look like this:

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)
)

因此对于这样的数据框：

So for a data frame like this:

| 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屋！