非时间戳列上的 Spark 结构化流 Window
Spark Strucutured Streaming Window on non-timestamp column
我正在获取以下形式的数据流：
+--+---------+---+----+
|id|timestamp|val|xxx |
+--+---------+---+----+
|1 |12:15:25 | 50| 1 |
|2 |12:15:25 | 30| 1 |
|3 |12:15:26 | 30| 2 |
|4 |12:15:27 | 50| 2 |
|5 |12:15:27 | 30| 3 |
|6 |12:15:27 | 60| 4 |
|7 |12:15:28 | 50| 5 |
|8 |12:15:30 | 60| 5 |
|9 |12:15:31 | 30| 6 |
|. |... |...|... |
我有兴趣将 window 操作应用到 xxx 列，就像 Spark Streaming 中可用的 window 时间戳操作一样，具有一些 window 大小和滑动步骤.
让下面的 groupBy 和 window 函数，lines 表示一个流式数据帧 window 大小：2 和滑动步长：1.
val c_windowed_count = lines.groupBy(
window($"xxx", "2", "1"), $"val").count().orderBy("xxx")
所以，输出应该如下：
+------+---+-----+
|window|val|count|
+------+---+-----+
|[1, 3]|50 | 2 |
|[1, 3]|30 | 2 |
|[2, 4]|30 | 2 |
|[2, 4]|50 | 1 |
|[3, 5]|30 | 1 |
|[3, 5]|60 | 1 |
|[4, 6]|60 | 2 |
|[4, 6]|50 | 1 |
|... |.. | .. |
我尝试使用 partitionBy，但 Spark Structured Streaming 不支持它。
我正在使用 Spark Structured Streaming 2.3.1。
谢谢！
目前无法使用 Spark Structured Streaming 以这种方式在非时间戳列上使用 windows。但是，您可以做的是 将 xxx 列转换为时间戳列 ，执行 groupBy 和 count，然后再转换回来。
from_unixtime 可用于将自 1970-01-01 以来的秒数转换为时间戳。使用 xxx 列作为秒，可以创建一个 假时间戳 以在 window 中使用：
lines.groupBy(window(from_unixtime($"xxx"), "2 seconds", "1 seconds"), $"val").count()
.withColumn("window", struct(unix_timestamp($"window.start"), unix_timestamp($"window.end")).as("window"))
.filter($"window.col1" =!= 0)
.orderBy($"window.col1")
上面，分组是在转换后的时间戳上完成的，下一行会将其转换回原来的数字。过滤器完成，因为前两行将是 window [0,2]（即仅在 xxx 等于 1 的行上）但可以跳过。
以上输入的结果输出：
+------+---+-----+
|window|val|count|
+------+---+-----+
| [1,3]| 50| 2|
| [1,3]| 30| 2|
| [2,4]| 30| 2|
| [2,4]| 50| 1|
| [3,5]| 30| 1|
| [3,5]| 60| 1|
| [4,6]| 60| 2|
| [4,6]| 50| 1|
| [5,7]| 30| 1|
| [5,7]| 60| 1|
| [5,7]| 50| 1|
| [6,8]| 30| 1|
+------+---+-----+
spark 2.2 中的新功能是 arbitrary-stateful-operations
一个用例是管理用户会话，一个 'user window'
scroll half way down this page to see an example
如果 Shaido 的巧妙解决方案对您有效，那么我建议您继续使用它。对于更复杂的需求，任意状态操作看起来像是要走的路。
我正在获取以下形式的数据流：
+--+---------+---+----+
|id|timestamp|val|xxx |
+--+---------+---+----+
|1 |12:15:25 | 50| 1 |
|2 |12:15:25 | 30| 1 |
|3 |12:15:26 | 30| 2 |
|4 |12:15:27 | 50| 2 |
|5 |12:15:27 | 30| 3 |
|6 |12:15:27 | 60| 4 |
|7 |12:15:28 | 50| 5 |
|8 |12:15:30 | 60| 5 |
|9 |12:15:31 | 30| 6 |
|. |... |...|... |
我有兴趣将 window 操作应用到 xxx 列，就像 Spark Streaming 中可用的 window 时间戳操作一样，具有一些 window 大小和滑动步骤.
让下面的 groupBy 和 window 函数，lines 表示一个流式数据帧 window 大小：2 和滑动步长：1.
val c_windowed_count = lines.groupBy(
window($"xxx", "2", "1"), $"val").count().orderBy("xxx")
所以，输出应该如下：
+------+---+-----+
|window|val|count|
+------+---+-----+
|[1, 3]|50 | 2 |
|[1, 3]|30 | 2 |
|[2, 4]|30 | 2 |
|[2, 4]|50 | 1 |
|[3, 5]|30 | 1 |
|[3, 5]|60 | 1 |
|[4, 6]|60 | 2 |
|[4, 6]|50 | 1 |
|... |.. | .. |
我尝试使用 partitionBy，但 Spark Structured Streaming 不支持它。
我正在使用 Spark Structured Streaming 2.3.1。
谢谢！
目前无法使用 Spark Structured Streaming 以这种方式在非时间戳列上使用 windows。但是，您可以做的是 将 xxx 列转换为时间戳列 ，执行 groupBy 和 count，然后再转换回来。
from_unixtime 可用于将自 1970-01-01 以来的秒数转换为时间戳。使用 xxx 列作为秒，可以创建一个 假时间戳 以在 window 中使用：
lines.groupBy(window(from_unixtime($"xxx"), "2 seconds", "1 seconds"), $"val").count()
.withColumn("window", struct(unix_timestamp($"window.start"), unix_timestamp($"window.end")).as("window"))
.filter($"window.col1" =!= 0)
.orderBy($"window.col1")
上面，分组是在转换后的时间戳上完成的，下一行会将其转换回原来的数字。过滤器完成，因为前两行将是 window [0,2]（即仅在 xxx 等于 1 的行上）但可以跳过。
以上输入的结果输出：
+------+---+-----+
|window|val|count|
+------+---+-----+
| [1,3]| 50| 2|
| [1,3]| 30| 2|
| [2,4]| 30| 2|
| [2,4]| 50| 1|
| [3,5]| 30| 1|
| [3,5]| 60| 1|
| [4,6]| 60| 2|
| [4,6]| 50| 1|
| [5,7]| 30| 1|
| [5,7]| 60| 1|
| [5,7]| 50| 1|
| [6,8]| 30| 1|
+------+---+-----+
spark 2.2 中的新功能是 arbitrary-stateful-operations
一个用例是管理用户会话，一个 'user window'
scroll half way down this page to see an example
如果 Shaido 的巧妙解决方案对您有效，那么我建议您继续使用它。对于更复杂的需求，任意状态操作看起来像是要走的路。