为什么我的水印不在我的Apache Flink Keyed Stream中前进？

Question

我目前正在将Apache Flink 1.13.2与Java一起用于流媒体应用程序。我正在使用没有窗口功能的键功能。尽管我的水印没有前进，但我已经实施了水印策略和autoWatermark Interval配置。

我通过使用Flink Web UI并在我的EventProcessor keyedProcessFunction中打印当前水印进行了仔细检查，但是水印会不断设置为很大的负数号码-9223372036854775808（最低水印）。

env.getConfig().setAutoWatermarkInterval(1000);

WatermarkStrategy<EventPayload> watermarkStrategy = WatermarkStrategy
        .<EventPayload>forMonotonousTimestamps()
        .withTimestampAssigner((event, timestamp) -> event.getTimestamp());

DataStream<EventPayload> deserialized = input
        .assignTimestampsAndWatermarks(watermarkStrategy)
        .flatMap(new Deserializer());

DataStream<EnrichedEventPayload> resultStream =
        AsyncDataStream.orderedWait(deserialized, new Enrichment(), 5, TimeUnit.SECONDS, 100);

DataStream<Session> eventsStream = resultStream
        .filter(EnrichedEventPayload::getIsEnriched)
        .keyBy(EnrichedEventPayload::getId)
        .process(new EventProcessor());

我什至尝试将WaterMarkStrategy添加到使用keyby的流中（并调整以匹配的类型），但仍然没有运气。

DataStream<Session> eventsStream = resultStream
        .filter(EnrichedEventPayload::getIsEnriched)
        .keyBy(EnrichedEventPayload::getId)
        .assignTimestampsAndWatermarks(watermarkStrategy)
        .process(new EventProcessor());

我还尝试使用自己的类实施WaterMarkStrategy并在OneVent函数上设置断点，以确保发布新的水印，尽管它仍然没有前进（并且任何相关的关联计时器没有开火）。

任何帮助将不胜感激！

Answer 1

如果水印策略的平行实例之一是空闲的（即，如果没有事件流过它），则会发生这种情况。在水印策略上使用withIdleness（...）选项将是解决此问题的一种方法。