状态保存 | 生而为人

状态保存的作用

将每次流计算的group结果，保存下来，用于下一次继续聚合。

具体执行内容是如何的？下次聚合会使用状态中的什么数据？

groupState.setTimeoutTimestamp()

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.

timeout晚于watermark不生效，早于watermark会报错？到底应该如何设置？

问题

flatmapgroupswithstate 什么时候会timeout
state-store-maintenance-task

参数调整

val spark = SparkSession
      .builder()
      .appName(appName)
      .config(sparkConf)
      .getOrCreate()
spark.sessionState.conf.setConfString("spark.sql.streaming.minBatchesToRetain", "5")

这个参数放到sparkConf中，无效。

状态保存的作用

问题

参考资料