Flink中的检查点事件时间水印

Question

我们正在接收来自多个独立数据源的事件，因此，到达我们Flink拓扑（通过Kafka）的数据将是无序的。

我们正在Flink拓扑中创建1分钟的事件时间窗口，并在源操作符处生成事件时间水印（当前事件时间-某些阈值（30秒））。

如果一些事件在设置的阈值之后到达，那么这些事件将被忽略（在我们的例子中这是可以的，因为属于该分钟的大多数事件都已经到达并在相应的窗口中得到处理）。

现在的问题是，如果程序崩溃（无论出于何种原因），然后从最后一个成功的检查点再次恢复，无序到达的事件将触发过去（已处理）窗口的执行（该窗口中只有一小部分事件）覆盖上一个检查点的结果。该窗口的计算。

如果Flink检查了事件时间水印，则不会发生此问题。

所以，我想知道是否有一种方法可以在Flink中强制执行事件时间水印的检查点...

Answer 1

我认为最简单的解决方案是在窗口操作符之后插入ProcessFunction。

ProcessFunction可通过其上下文对象访问当前水印，并可将其存储在联合运算符状态。如果出现故障，ProcessFunction将水印从其状态恢复，并过滤时间戳小于水印的所有记录（时间戳也可通过上下文obejct访问）。

Answer 2

虽然这是一个老问题，但我也有同样的问题。应用程序正在重新启动，具有事件时间窗口的连接函数不再触发，因为来自其中一个流的事件在崩溃前完成。连接可以恢复状态，但是由于其中一个流不再有水印，事件在重启后永远不会连接。

我找到的解决方案是在源操作符之后为最新的水印创建一个检查点。由于没有UDF来保存水印的快照，我必须创建自己的操作符，该操作符不会更改事件（标识函数），并将最新的水印保存为其状态。当Flink从崩溃中恢复时，WatermarkStreamOperator。InitializeEstate（）发出列表状态上的最后一个水印检查点

public class WatermarkStreamOperator<IN> extends AbstractUdfStreamOperator<IN, WatermarkFunction<IN>>
        implements OneInputStreamOperator<IN, IN> {
    private static final long serialVersionUID = 1L;
    private ListState<Long> latestWatermark;
    public WatermarkStreamOperator(WatermarkFunction<IN> mapper) {
        super(mapper);
        chainingStrategy = ChainingStrategy.ALWAYS;
    }
    @Override
    public void initializeState(StateInitializationContext context) throws Exception { System.out.println("WatermarkStreamOperator.initializeState");
        super.initializeState(context);
        ListStateDescriptor<Long> descriptor = new ListStateDescriptor<>("latest-watermark", Long.class);
        latestWatermark = context.getOperatorStateStore().getListState(descriptor);
        List<Long> watermarkList = new ArrayList<>();
        latestWatermark.get().forEach(watermarkList::add);
        Long maxWatermark = watermarkList.stream().max(Long::compare).orElse(0L);
        if (!maxWatermark.equals(Long.valueOf(0l))) {
            System.out.println("watermarkList recovered max: " + maxWatermark);
            processWatermark(new Watermark(maxWatermark));
        }
    }
    @Override
    public void processElement(StreamRecord<IN> element) throws Exception {
        output.collect(element);
    }
    @Override
    public void processWatermark(Watermark mark) throws Exception {
        System.out.println("processing watermark: " + mark.getTimestamp()); latestWatermark.update(Arrays.asList(mark.getTimestamp()));
        super.processWatermark(mark);
    }
}

以及操作员的标识UDF：

public interface WatermarkFunction<T> extends Function, Serializable {
    T process(T value) throws Exception;
}

最后，我使用。转换（）调用我的Watermark Stream算子与MyTupleWatermark Func。

DataStream<Tuple2<String, Integer>> dataStream = env
                .addSource(new MySource(sentence))
                .transform("myStatefulWatermarkOperator",
                        TypeInformation.of(String.class),
                        new WatermarkStreamOperator<>(new MyTupleWatermarkFunc()))
                ...
                ...
    public class MyTupleWatermarkFunc implements WatermarkFunction<String> {
        private static final long serialVersionUID = 1L;
        @Override
        public String process(String value) throws Exception {
            return value;
        }
    }

下面是我为这个https://github.com/felipegutierrez/explore-flink/blob/master/docker/ops-playground-image/java/explore-flink/src/test/java/org/sense/flink/examples/stream/operator/watermark/WatermarkStreamOperatorTest.java创建的单元和集成测试