Apache Flink 1.6.0 - StateTtlConfig 和 ListState

Question

我正在使用 Apache Flink 1.6.0 实现概念验证流处理系统，并且正在 ListState 中存储按键分区的已接收事件列表。（不要担心我为什么要这样做，在这里和我一起工作。）我在相应的 ListStateDescriptor 上设置了 StateTtlConfig。 Per the documentation:

"All state collection types support per-entry TTLs. This means that list elements and map entries expire independently."
"Currently, expired values are only removed when they are read out explicitly, e.g. by calling ValueState.value()."

问题 1

以下哪项构成对 ListState 的阅读：

请求迭代器但不使用它 - myListState.get();。
实际使用迭代器 - for (MyItem i : myListState.get()) { ... }

问题二

"per-entry TTL" 实际上 是什么意思？具体来说，我要问的是以下内容：

假设我有一个 ListState<Character> 的特定实例。描述符的 TTL 为 10 秒。我插入一个'a'。两秒钟后，我插入 'b'。九秒后我插入 'c'。如果我遍历此 ListState，将返回哪些项目？

换句话说：

ListState<Character> ls = getRuntimeContext().getListState(myDescriptor);

ls.add('a');

// ...two seconds later...
ls.add('b');

// ...nine seconds later...
ls.add('c');

// Does this iterate over 'a', 'b', 'c'
// or just 'b' and 'c'?
for (Character myChar : ls.get()) { ... }

Answer 1

回答 1

答案是 1。对于 ListState，修剪是针对 myListState.get();。

答案 2

"per-entry TTL" 表示超时应用于单个条目而不是整个集合。对于您的示例，假设在读取时自插入 a 后经过 10 秒，它将遍历 b 和 c。 a 在这种情况下将被修剪。

Apache Flink 1.6.0 - StateTtlConfig 和 ListState

Apache Flink 1.6.0 - StateTtlConfig and ListState

events

stateful

ttl

apache-flink