使用 Spark Groupby 聚合查找最常见的值和相应的计数

Question

我正在尝试使用 Spark (Scala) 数据帧对模式和相应计数进行分组聚合。

例如，

假设我们有以下数据框：

Category   Color   Number   Letter      
1        Red         4        A
1        Yellow      Null     B
3        Green       8        C
2        Blue        Null     A
1        Green       9        A
3        Green       8        B
3        Yellow      Null     C
2        Blue        9        B
3        Blue        8        B
1        Blue        Null     Null
1        Red         7        C
2        Green       Null     C
1        Yellow      7        Null
3        Red         Null     B

现在我们要按类别分组，然后按颜色分组，然后求分组的大小、非空数的计数、数字的总大小、数字的均值、数字的众数，以及相应的模式计数。对于字母，我想要非空值的计数以及相应的模式和模式计数（不是意思，因为这是一个字符串）。

所以理想情况下输出是：

Category     Color     CountNumber(Non-Nulls)   Size   MeanNumber  ModeNumber ModeCountNumber   CountLetter(Non-Nulls)  ModeLetter   ModeCountLetter
1            Red       2                        2      5.5         4 (or 7) 
1            Yellow    1                        2      7           7     
1            Green     1                        1      9           9       
1            Blue      1                        1      -           -       
2            Blue      1                        2      9           9      etc 
2            Green     -                        1      -           -       
3            Green     2                        2      8           8       
3            Yellow    -                        1      -           -       
3            Blue      1                        1      8           8       
3            Red       -                        1      -           -

这对于计数和均值来说很容易做到，但对于其他一切就比较棘手了。如有任何建议，我们将不胜感激。

谢谢。

Answer 1

据我所知——没有简单的方法来计算模式——你必须计算每个值的出现次数，然后将结果与该结果的最大值（每个键）连接起来。其余的计算相当简单：

// count occurrences of each number in its category and color
val numberCounts = df.groupBy("Category", "Color", "Number").count().cache()

// compute modes for Number - joining counts with the maximum count per category and color:
val modeNumbers = numberCounts.as("base").join(numberCounts.groupBy("Category", "Color").agg(max("count") as "_max").as("max"),
  $"base.Category" === $"max.Category" and
  $"base.Color" === $"max.Color" and
  $"base.count" === $"max._max")
  .select($"base.Category", $"base.Color", $"base.Number", $"_max")
  .groupBy("Category", "Color")
  .agg(first($"Number", ignoreNulls = true) as "ModeNumber", first("_max") as "ModeCountNumber")
  .where($"ModeNumber".isNotNull)

// now compute Size, Count and Mean (simple) and join to add Mode:
val result = df.groupBy("Category", "Color").agg(
  count("Color") as "Size", // counting a key column -> includes nulls
  count("Number") as "CountNumber", // does not include nulls
  mean("Number") as "MeanNumber"
).join(modeNumbers, Seq("Category", "Color"), "left")

result.show()
// +--------+------+----+-----------+----------+----------+---------------+
// |Category| Color|Size|CountNumber|MeanNumber|ModeNumber|ModeCountNumber|
// +--------+------+----+-----------+----------+----------+---------------+
// |       3|Yellow|   1|          0|      null|      null|           null|
// |       1| Green|   1|          1|       9.0|         9|              1|
// |       1|   Red|   2|          2|       5.5|         7|              1|
// |       2| Green|   1|          0|      null|      null|           null|
// |       3|  Blue|   1|          1|       8.0|         8|              1|
// |       1|Yellow|   2|          1|       7.0|         7|              1|
// |       2|  Blue|   2|          1|       9.0|         9|              1|
// |       3| Green|   2|          2|       8.0|         8|              2|
// |       1|  Blue|   1|          0|      null|      null|           null|
// |       3|   Red|   1|          0|      null|      null|           null|
// +--------+------+----+-----------+----------+----------+---------------+

如您所想 - 这可能会很慢，因为它有 4 个 groupBy 和两个 join - 都需要洗牌...

至于 Letter 列统计信息 - 恐怕您必须单独为该列重复此操作并添加另一个连接。

使用 Spark Groupby 聚合查找最常见的值和相应的计数

Find Most Common Value and Corresponding Count Using Spark Groupby Aggregates

scala

group-by

aggregation

apache-spark

apache-spark-sql