如何在 Apache Flink 中使用 TTL 使键控状态过期？

Question

我有这样的管道：

 env.addSource(kafkaConsumer)
            .keyBy { value -> value.f0 }
            .window(EventTimeSessionWindows.withGap(Time.minutes(2)))
            .reduce(::reduceRecord)
            .addSink(kafkaProducer)

我想用 TTL 使键控数据过期。

一些博客文章指出我需要一个ValueStateDescriptor。 我做了一个这样的：

val desc = ValueStateDescriptor("val state", MyKey::class.java)
desc.enableTimeToLive(ttlConfig)

但是我如何将这个描述符实际应用到我的管道中，以便它实际执行 TTL 到期？

Answer 1

您描述的管道不使用任何可以从设置状态 TTL 中受益的键控状态。管道中唯一的键控状态是会话窗口的内容，并且随着会话关闭，该状态将被尽快清除。 （此外，由于您使用的是 reduce 函数，因此该状态仅包含每个键的一个值。）

在大多数情况下，过期状态仅与您明确创建的状态相关，在这种情况下，您可以随时访问状态描述符并可以将其配置为使用状态 TTL。 Flink SQL 确实会代表您创建可能不会自动过期的状态，在这种情况下，您需要使用 Idle State Retention Time 来配置它。 CEP 库还代表您创建状态，在这种情况下，您应该确保您的模式最终匹配​​或超时。