scala - Spark Bucketizer - 即使没有元素也显示所有桶

Question

我有一个事件列表，如下所示从 spark 数据帧生成。我在 Scala 中使用 Spark 2.2.0。

val events = df.select($"event", hour($"time") as "hour", to_date($"time", "yyyy-MM-dd") as "day")


+-----+-----+---------+
|事件|时间|天|
+-----+-----+---------+
|事件1| 18|2015-02-05|
|事件1| 17|2015-02-19|
|事件5| 18|2015-02-02|
|事件5| 19|2015-02-02|
|事件1| 1|2015-03-17|
|事件1| 0|2015-02-03|
|事件1| 20|2015-02-02|
|事件1| 22|2015-02-02|
|事件1| 23|2015-02-02|
|事件1| 18|2015-02-09|
|事件1| 19|2015-02-09|
|事件1| 21|2015-02-09|
|事件1| 21|2015-04-06|
|事件1| 23|2015-02-09|
|事件1| 20|2015-02-16|
|事件2| 19|2015-02-12|
|事件3| 18|2015-02-18|
|事件1| 22|2015-02-16|
|事件2| 17|2015-02-04|
|事件1| 23|2015-02-16|
+-----+-----+---------+
只显示前 20 行


我需要创建每小时存储桶并计算每小时发生的事件数。所以我的方法是创建存储桶(其中 24 个)并计算其特定时段内的事件，如下所示。

val splits = (0 to 24).map(_ * 1.0).toArray
val bucketizer = new Bucketizer()
    .setInputCol("hour")
    .setOutputCol("bucket")
    .setSplits(splits)

val bucket = bucketizer.transform(events)

val result = bucket.groupBy($"day", $"bucket").agg(count($"event").as("count")).orderBy(asc("bucket"))

result.filter($"day" === "2015-05-21").orderBy(asc("bucket")).show()

上面代码的结果是


+------------+------+-----+
|天|桶|计数|
+------------+------+-----+
|2015-05-21| 0.0| 1|
|2015-05-21| 2.0| 1|
|2015-05-21| 11.0| 1|
|2015-05-21| 17.0| 1|
|2015-05-21| 18.0| 4|
|2015-05-21| 19.0| 4|
|2015-05-21| 21.0| 1|
|2015-05-21| 22.0| 3|
|2015-05-21| 23.0| 1|
+------------+------+-----+


这是正确的。然而，我期望的输出是这样的:


+------------+------+-----+
|天|桶|计数|
+------------+------+-----+
|2015-05-21| 0.0| 1|
|2015-05-21| 1.0| 0|
|2015-05-21| 2.0| 1|
|2015-05-21| 3.0| 0|
|2015-05-21| 4.0| 0|
|2015-05-21| 5.0| 0|
:
:
|2015-05-21| 11.0| 1|
|2015-05-21| 12.0| 0|
|2015-05-21| 13.0| 0|
:
:
|2015-05-21| 17.0| 1|
|2015-05-21| 18.0| 4|
|2015-05-21| 19.0| 4|
|2015-05-21| 20.0| 0|
|2015-05-21| 21.0| 1|
|2015-05-21| 22.0| 3|
|2015-05-21| 23.0| 1|
+------------+------+-----+


基本上，没有事件的容器(桶)应该填充 0。知道如何实现吗？

谢谢!

Answer 1

这是我目前没有使用 Bucketizer 的解决方案(我承认不是很漂亮)

val events = df.select($"event", hour($"time") as "hour", to_date($"time", "yyyy-MM-dd") as "day")

val left = (0 to 24).toDF.withColumnRenamed("value", "hour")
val right = or_counts.filter($"day" === "2015-05-21").groupBy($"hour").agg(count("event").as("count")).orderBy(asc("hour"))

left.join(right, Seq("hour"), "left_outer").na.fill(0, Seq("count")).show()

此代码返回如下所示的内容:


+----+-----+
|小时|计数|
+----+-----+
| 0| 1|
| 1| 0|
| 2| 1|
| 3| 0|
| 4| 0|
| 5| 0|
| 6| 0|
| 7| 0|
| 8| 0|
| 9| 0|
| 10| 0|
| 11| 1|
| 12| 0|
| 13| 0|
| 14| 0|
| 15| 0|
| 16| 0|
| 17| 1|
| 18| 4|
| 19| 4|
| 20| 0|
| 21| 1|
| 22| 3|
| 23| 1|
| 24| 0|
+----+-----+
这是我的预期结果。如果有人能提出更好的解决方案，我会接受这个答案。

谢谢!