mapGroupsWithState 中的火花流理解超时设置

Question

在使用 mapGroupsWithState 进行 spark 结构化流式传输时，我正在努力理解超时设置。

下面的链接有非常详细的规范，但我不确定我是否理解正确，尤其是GroupState.setTimeoutTimeStamp() 选项。将状态到期设置为与事件时间相关的含义。 https://spark.apache.org/docs/3.0.0-preview/api/scala/org/apache/spark/sql/streaming/GroupState.html

我在这里复制了它们：

With EventTimeTimeout, the user also has to specify the the the event time watermark in the query using Dataset.withWatermark(). 

With this setting, data that is older than the watermark are filtered out. 
The timeout can be set for a group by setting a timeout timestamp usingGroupState.setTimeoutTimestamp(), and the timeout would occur when the watermark advances beyond the set timestamp. 

You can control the timeout delay by two parameters - watermark delay and an additional duration beyond the timestamp in the event (which is guaranteed to be newer than watermark due to the filtering). 

Guarantees provided by this timeout are as follows:
Timeout will never be occur before watermark has exceeded the set timeout.
Similar to processing time timeouts, there is a no strict upper bound on the delay when the timeout actually occurs. The watermark can advance only when there is data in the stream, and the event time of the data has actually advanced.

问题 1： 这句and the timeout would occur when the watermark advances beyond the set timestamp中的timestamp是什么？它是绝对时间，还是该州当前事件时间的相对持续时间？我知道我可以通过 ```

删除状态来使其过期

例如假设我有一些如下的数据状态，when 会通过在what settings 中设置what value 来过期吗？

+-------+-----------+-------------------+
|expired|something  |          timestamp|
+-------+-----------+-------------------+
|  false|   someKey |2020-08-02 22:02:00|
+-------+-----------+-------------------+

问题 2： 看了Data that is older than the watermark are filtered out这句话，我明白迟到的数据从kafka读取后会被忽略，这样对吗？

问题原因 如果不了解这些，我就无法真正将它们应用到用例中。含义何时使用GroupState.setTimeoutDuration()，何时使用GroupState.setTimeoutTimestamp()

非常感谢。

ps。我也尝试阅读以下内容

-  https://www.waitingforcode.com/apache-spark-structured-streaming/stateful-transformations-mapgroupswithstate/read
(confused me, did not understand)
- https://databricks.com/blog/2017/10/17/arbitrary-stateful-processing-in-apache-sparks-structured-streaming.html
(did not say a lot of it for my interest)

Answer 1

句子and the timeout would occur when the watermark advances beyond the set timestamp中的timestamp是什么？

这是GroupState.setTimeoutTimestamp()设置的时间戳。

是绝对时间还是相对于状态当前事件时间的相对持续时间？

这是基于当前批处理窗口的相对时间（不是持续时间）。

假设我有一些数据状态（列timestamp=2020-08-02 22:02:00），通过在什么设置中设置什么值它什么时候到期？

假设您的接收器查询有一个定义为 5 分钟的处理触发器（由trigger() 设置）。另外，让我们假设您在应用groupByKey 和mapGroupsWithState 之前使用了水印。我了解您希望使用基于 事件时间 的超时（而不是 处理时间，因此您的查询将类似于：

ds.withWatermark("timestamp", "10 minutes")
  .groupByKey(...) // declare your key
  .mapGroupsWithState(
    GroupStateTimeout.EventTimeTimeout)(
    ...) // your custom update logic

现在，这取决于您如何使用“自定义更新逻辑”设置 TimeoutTimestamp。在您的自定义更新逻辑中的某处，您需要调用

state.setTimeoutTimestamp()

此方法有四种不同的签名，值得仔细阅读它们的文档。由于我们在 (withWatermark) 中设置了水印，我们实际上可以利用这段时间。作为一般规则：将超时时间戳（由state.setTimeoutTimestamp() 设置）设置为大于当前水印的值很重要。为了继续我们的示例，我们添加一小时，如下所示：

state.setTimeoutTimestamp(state.getCurrentWatermarkMs, "1 hour")

总而言之，您的消息可以在22:00:00 和22:15:00 之间到达您的流，如果该消息是密钥的最后一条消息，它将在您的 GroupState 中超时23:15:00。

问题2：读Data that is older than the watermark are filtered out这句话，我理解迟到的数据从kafka读取后被忽略了，对吗？

是的，这是正确的。对于批处理时间间隔 22:00:00 - 22:05:00，所有具有事件时间（由列 timestamp 定义）的消息都将晚于声明的 10 分钟水印（意味着晚于 22:15:00）到达在您的查询中无论如何都会被忽略，并且不会在您的“自定义更新逻辑”中进行处理。