apache-spark - 在 Spark 结构化流中按键加水印

Question

我有来自 IoT 设备的 Kafka 数据。由于网络拥塞、设备超出范围等原因，这些设备的传感器数据的时间戳往往不同步。

我们必须编写流式传输作业，以便在一段时间内为每个设备独立聚合传感器值。通过带水印操作的 groupby，我们丢失了所有落后于具有最新时间戳的设备的数据。

有什么方法可以根据设备的最新时间戳而不是所有设备的最新时间戳将水印分别应用到每个设备？

我们不能保持很大的滞后，因为设备可能会超出范围几天。由于设备数量很多，我们无法针对每台设备运行单独的查询。

使用 flatMapGroupsWithState 可以实现吗？或者这是 Spark Structured Streaming 根本无法实现的？

Answer 1

我认为您可以在处理时间戳(即您在 Spark 作业中处理数据的时间)上应用水印，而不是按事件时间戳添加水印(这可能会像您所说的那样滞后)。我在最近的一个项目中遇到了一个非常相似的问题，我就是这样解决的。

例子:

val dfWithWatermark = df
  .withColumn("processingTimestamp", current_timestamp())
  .withWatermark("processingTimestamp", "1 day")

// ... use dfWithWatermark to do aggregations etc

无论您收到的数据的时间戳是什么，这都会使您的 IoT 数据保持超过 1 天的状态。

当然，这有一些限制，例如，如果有设备以大于水印的间隔发送数据。但要为此找出解决方案，您必须更具体地满足您的要求。

通过使用 flatMapGroupsWithState，您可以非常特定地保持状态，但我认为在您的情况下没有必要。

编辑:但是，如果您决定使用 flatMapGroupsWithState，您可以通过以不同的时间间隔调用 state.setTimeoutDuration() 来为每个设备组使用不同的超时，具体取决于您处理的设备类型。这样您就可以非常具体地保留状态。