使用 Spark Structured Streaming 对传感器数据进行超时分组

Question

我们让传感器每天多次启动并随机运行。来自传感器的数据被发送到 Kafka 主题，并由 Spark 结构化流 API 使用并存储到 Delta Lake。现在我们必须识别每个传感器的会话并将其存储在由 device_id 和 sensor_id 分区的不同 Delta Lake 表中。

我尝试使用带有水印的 Spark Structured Streaming，但效果不佳。

stream2 = spark.readStream.format('delta')
             .load('<FIRST_DELTA_LAKE_TABLE>')
             .select('device_id', 'json', 'time')
             .withWatermark('timestamp', '10 minutes')
             .groupBy('device_id').agg(F.min('time').alias('min_time'), F.max('time').alias('max_time')))
             .writeStream
             .format("delta")
stream2.start("<SESSIONS_TABLE>")

我们的想法是让第二个表从传入数据中识别会话并保存每个会话和设备的开始时间和结束时间。流作业运行，没有任何内容写入 Sessions 增量表。

我们将不胜感激。

Answer 1

默认情况下，当您编写流时，它默认使用append 模式（请参阅doc）。而在这种模式下，当你使用水印时，只有越过水印后才会输出数据，所以至少要延迟10分钟才能开始看到输出中的数据。

但我认为主要问题是您正在运行“全局”聚合，没有定义窗口或类似的东西。通常人们使用flatMapGroupWithState进行会话检测，类似于下面的blog post。