Spark 数据帧 groupBy 和进一步计数聚合答案

【问题标题】：Spark dataframe groupBy and further count aggregationsSpark 数据帧 groupBy 和进一步计数聚合
【发布时间】：2018-09-03 01:44:09
【问题描述】：

数据框示例：

col_1, col_2
aaa, 1
aaa, 0
bbb, 1
bbb, 1
bbb, 1

我想要一个包含 3 列的结果 df：col_1、总行数以及 col_2 === 1 的行数。

我试过了

df.groupBy($"col_1")
   .agg(count($"col_2" === 1).as("delayed"), count(lit(1)) as "total").show(100)

为什么总计算正确，但延迟计算不正确？

【问题讨论】：

我知道另一种实现这一目标的方法，但我的问题是为什么我使用这种方法会得到错误的结果

标签： scala apache-spark dataframe aggregation

【解决方案1】：

问题

当你使用count($"col_2" === 1).as("delayed")时

引用count function

public static Column count(Column e) Aggregate function: returns the number of items in a group. Parameters: e - (undocumented) Returns: (undocumented) Since: 1.3.0

而不是scala count function

def count(p : (A) => Boolean) : Int Count the number of elements in the list which satisfy a predicate. Parameters p - the predicate for which to count Returns the number of elements satisfying the predicate p.

因此count($"col_2" === 1) 中的条件即$"col_2" === 1 不被视为true 或false，而是被视为列。 因此计数函数只计算列$"col_2" === 1。

我希望解释清楚易懂。

解决方案

您应该使用when 函数将值更改为1 或0 并使用sum 函数作为计数。

sum(when($"col_2" === 1, 1).otherwise(0)).as("delayed")

如果 col_2 列总是 0 或 1 那么您只能使用 sum

sum($"col_2").as("delayed")

【讨论】：

【解决方案2】：

$"col2"===1 列的元素数量仍与$"col2" 相同，它们只是true 或false。相反，您想转换为整数并求和。（当然，如果col2的值一直是1或者0，直接求和就可以了。）

我认为你必须定义一个 udf 来将布尔值转换为整数：

val toInt = udf((x: Boolean) => if(x) 1 else 0)

然后（我没有为我的专栏命名）：

scala> df.groupBy($"_1").agg(sum(toInt($"_2"===1)), count($"_2")).show()
+---+------------------+---------+
| _1|sum(UDF((_2 = 1)))|count(_2)|
+---+------------------+---------+
|aaa|                 1|        2|
|bbb|                 3|        3|
+---+------------------+---------+

【讨论】：

有趣的是这两个计数不相等
我不完全确定你的意思。我认为您可能将“count”和“count if”混为一谈，而count 仅计数。
稍后我会用具体数字添加更多细节