Flink keyedstream生成具有相同密钥和窗口时间戳的重复结果

Question

这是我的Flink工作流程：

DataStream<FlinkEvent> events = env.addSource( consumer ).flatMap(...).assignTimestampsAndWatermarks( new EventTsExtractor() );
DataStream<SessionStatEvent> sessionEvents = events.keyBy( 
    new KeySelector<FlinkEvent, Tuple2<String, String> >()
            {
                @Override
                public Tuple2<String, String> getKey( FlinkEvent value ) throws Exception {
                    return(Tuple2.of( value.getF0(), value.getSessionID ) );
                }
            } )
      .window( TumblingEventTimeWindows.of( Time.minutes( 2 ) ) )
      .allowedLateness( Time.seconds( 10 ) )
      .aggregate( new SessionStatAggregator(), new SessionStatProcessor() );
/* ... */
sessionEvents.addSink( esSinkBuilder.build() );

首先我遇到了

java.lang.Exception： org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException： Can not forward element to next operator

在< code>flatMap运算符中，任务继续重新启动。我观察到许多重复的结果，相同的键和窗口时间戳具有不同的值。

Q1：我猜重复是因为下游操作员在作业重新启动后重复使用消息。我说得对吗？在修复了异常inchainedOperatorException问题后，我重新提交了作业。我又在第一个时间窗口中观察到了重复。在那之后，工作似乎进行得很顺利（每个键一个时间窗口中有一个结果）。

Q2:复制品是从哪里来的？

Answer 1

这就是Flink实现精确一次语义学的方法。如果失败，Flink会重放上次成功检查点的事件。重要的是要注意，精确一次意味着影响状态一次，而不是处理/发布事件一次。

回答问题1：是的，每次重新启动都会导致一遍又一遍地处理相同的消息 回答问题2：错误修复后的第一个窗口再次处理这些消息;然后一切都恢复正常。

Answer 2

...一个窗口的每个键应该有一个结果

这不(完全)正确。由于允许的延迟，任何延迟事件(在允许的延迟时间内)将导致相关窗口的延迟(或者换句话说，额外)触发。使用默认的EventTimeTrigger(您似乎正在使用它)，每个延迟事件都会导致一个额外的窗口触发，并且会发出一个更新的窗口结果。