Apache flink-早期触发窗口实现问题-收到重复元素

Question

我很难理解flink窗口原则，如果你能给我指明正确的方向，我将非常高兴。

我的目的是计算一个时间间隔内重复发生的事件的数量，如果重复发生的事件的数量大于阈值，则生成警报事件。

我正在使用事件时语义和自定义触发器的问题/疑问。

您可以在GIST中找到实际的实现:https://gist.github.com/simpleusr/7C56D4384F6FC9F0A61860A680BB5F36

我使用键控状态来跟踪窗口中的元素计数EncounteredelementsCountState

问题是，我不得不在不明白原因的情况下插入下面的检查代码。因为以前收集的元素再次提供给onelement方法：

if (ctx.getCurrentWatermark() < 0) {
            logger.debug(String.format("onElement processing skipped for eventId : %s for watermark: %s ", element.getEventId(), ctx.getCurrentWatermark()));
            return TriggerResult.CONTINUE;
        }

我想不出原因。我看到的是，当发生这种情况时，水印值为(ctx.getCurrentWatermark())long.min_value（这导致了上面的检查）。怎么会这样呢？

这种检查似乎避免了重复的早期事件生成，但我不知道为什么会发生这种情况，这种变通方法是否合适。

你能告诉我为什么相同的元素在窗口中被处理两次吗？

另一个问题是关于键控状态的用法。此实现在释放窗口后是否泄漏任何状态？我正在尝试清除触发器的clear方法中所有使用的状态，但这是否足够？

问候。

Answer 1

每个任务都将currentWatermark初始化为long.min_value，在该任务的所有输入流中都有较大的水印到达之前，该值保持为currentWatermark的本地值。希望知道这将帮助你更好地理解发生了什么。

就其价值而言，通常使用ProcessFunction实现这种逻辑比使用Window API更简单。