是否强制清除窗口末尾的窗口状态对象

Question

我使用window API将数据划分为1小时的窗口。在每个窗口中，我使用Value状态为每个窗口存储一个布尔值。

.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Event>(Time.days(1)) {
@Override
public long extractTimestamp(Event element) {
return element.timestamp;
}
})
// Partition by user
.keyBy(new KeySelector<Event, Tuple2<Long, String>>() {
@Override
public Tuple2<Long, String> getKey(Event value) {
return Tuple2.of(value.userGroup, value.userName);
}
})
.window(TumblingEventTimeWindows.of(Time.minutes(60), Time.minutes(0)))
.allowedLateness(Time.days(1))
.trigger(new WindowTrigger<>(EVENTS_THRESHOLD))
.aggregate(new WindowAggregator(), new WindowProcessor())
.print();

public class WindowProcessor extends ProcessWindowFunction<Long, String, Tuple2<Long, String>, TimeWindow> {
private final ValueStateDescriptor<Boolean> windowAlertedDescriptor = new ValueStateDescriptor<>("windowAlerted", Boolean.class);
@Override
public void process(Tuple2<Long, String> key, Context context, Iterable<Long> elements, Collector<String> out) throws Exception {
long currentDownloadsCount = elements.iterator().next();
long windowStart = context.window().getStart();
long windowEnd = context.window().getEnd();
ValueState<Boolean> windowAlertedState = context.windowState().getState(windowAlertedDescriptor);
if (BooleanUtils.isTrue(windowAlertedState.value())) {
return;
}

我必须打电话给"；clear(("；清除窗口状态数据的方法？我认为，因为Flink处理窗口的创建和清除，所以它在清除窗口时也应该处理状态清理。

根据这里的答案，如何在处理键控窗口后立即清除状态？一旦窗口启动，窗口就会自动清除其状态。

但是Flink文档明确指出，应该调用clear方法来删除窗口状态https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/operators/windows.html#using-处理窗口函数中的每个窗口状态

Answer 1

窗口API中涉及的各种类在多个位置保持状态：

• 分配给每个Window的流记录列表
• Trigger可以是有状态的(例如CountTrigger(
• 每个窗口状态(在ProcessWindowFunction.Context中(
• 全局状态(也在ProcessWindowFunction.Context中(

清除窗口时，Flink会自动清除前两个(窗口内容和触发器状态(。清除窗口时，Flink还会调用ProcessWindowFunction上的clear方法，此时应该清除在KeyedStateStore windowState()中创建的每个窗口的状态。

另一方面，KeyedStateStore globalState()的目的是从一个窗口记住另一个窗口的内容，所以您不会清除它。但是，如果您有一个无边界的键空间，则应该注意清理过时键的全局窗口状态。唯一的方法是在全局状态的状态描述符上指定状态TTL。