【发布时间】：2023-04-05 19:36:01
【问题描述】：
我们正在读取 Kinesis 并写入 parquet，我们使用 StateSpec<ValueState<Boolean>> 来避免在从最后一个保存点优雅地停止和重新启动我们的管道后重复处理记录。
我们看到一些记录被重复，因为它们在后续重新启动时最终落在不同的任务管理器上，我们使用StateSpec<ValueState<Boolean>> 来存储有关已处理记录的状态信息并避免重复。
我们正在处理如何在每次特定时间清除状态，而不会丢失最近处理的记录（如果在即将到来的停止中需要它们）。 （即，我们需要该类上的 TTL 之类的东西）。
我们考虑了一个计时器，它可以每隔一定时间清除状态，但这不符合我们的要求，因为我们需要保留最近处理的记录。
我们读到here，使用事件时间处理会在窗口到期后自动清除状态信息，我们想知道这是否符合我们使用StateSpec 类的要求。
否则，是否有一个类来存储具有某种 TTL 来实现此功能的状态？
我们现在拥有的是这段代码，它检查元素是否已经处理，以及一个每隔一定时间清除状态的方法
@StateId("keyPreserved")
private final StateSpec<ValueState<Boolean>> keyPreserved = StateSpecs.value(BooleanCoder.of());
@TimerId("resetStateTimer")
private final TimerSpec resetStateTimer = TimerSpecs.timer(TimeDomain.PROCESSING_TIME);
public void processElement(ProcessContext context,
@TimerId("resetStateTimer") Timer resetStateTimer,
@StateId("keyPreserved") ValueState<Boolean> keyPreservedState) {
if (!firstNonNull(keyPreservedState.read(), false)) {
T message = context.element().getValue();
//Process element here
keyPreservedState.write(true);
}
}
@OnTimer("resetStateTimer")
public void onResetStateTimer(OnTimerContext context,
@StateId("keyPreserved") ValueState<Boolean> keyPreservedState) {
keyPreservedState.clear();
}
【问题讨论】：
-
example under Event-Time-Timers 可以用来实现
using event time processing automatically clears State information after a window expires。您的示例代码似乎缺少如何配置计时器，它应该是 example under Processing-Time-Timers。 -
是的，我们关注那个帖子，我们认为对 ValueState 对象调用 clear() 会清除整个缓存。然后我们发现它只会清除上下文中的元素。如果我们很好地理解了最后一个行为（仅清除 1 个元素），应该会找到 TTL，而不会清除整个缓存。但我们试过了，我们仍然有重复的记录
-
另一方面，我们现在没有使用事件时间，我们正在使用处理时间。问题是当窗口到期时状态是否会自动清除，并且只有与该窗口相关的状态。如果我们使用您提到的方法并在到期后的某个时间对缓存进行 clear() 处理，那么我们处于相同的场景中（处理清除整个缓存 - 我们不想要它 - 或者可能不是？）
标签： apache-flink google-cloud-dataflow apache-beam amazon-kinesis