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

Question

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

.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-per-windown-state-in-processwindowfunction

Answer 1

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

• 分配给每个窗口
的流记录列表 触发器可以是有状态的（例如 CountTrigger)
• 每窗口状态(ProcessWindowFunction.Context)
• 全局状态（也在ProcessWindowFunction.Context)

前两个（窗口内容和触发器状态）在清除窗口时由Flink自动清除。清除窗口时，Flink还调用ProcessWindowFunction上的clear方法，并且您应该清除您当时在KeyedStateStoreWindowState()中创建的任何每个窗口状态。

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