Flink 设置 5 分钟 Watermark 指南

在 Flink 中，Watermark 用于控制数据处理的延迟，确保流处理的准确性和可靠性。以下步骤展示如何设置 5 分钟的 Watermark：

1. 自定义 Watermark 类

public class MyWatermark implements AssignerWithPeriodicWatermarks<MyEvent> {
    private final long maxOutOfOrderness = 5 * 60 * 1000; // 5 分钟
    private long currentMaxTimestamp;

    @Override
    public long extractTimestamp(MyEvent event, long previousElementTimestamp) {
        long timestamp = event.getTimestamp();
        currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
        return timestamp;
    }

    @Nullable
    @Override
    public Watermark getCurrentWatermark() {
        // 返回当前最大时间戳减去 5 分钟的 Watermark
        return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
    }
}

在这个例子中，我们创建了一个名为 MyWatermark 的类，并实现了 AssignerWithPeriodicWatermarks 接口。在 extractTimestamp 方法中，我们从事件中抽取时间戳，并更新当前最大时间戳。在 getCurrentWatermark 方法中，我们返回当前最大时间戳减去 5 分钟的 Watermark。

2. 应用 Watermark

DataStream<MyEvent> events = ...;

DataStream<MyEvent> withWatermarks = events
    .assignTimestampsAndWatermarks(new MyWatermark());

这将对 events 数据流应用 MyWatermark 实现的 Watermark。注意，我们需要在 assignTimestampsAndWatermarks 方法中传递一个 MyWatermark 实例。

总结

通过自定义 Watermark 类和应用 Watermark，我们可以控制数据处理的延迟，确保 Flink 流处理的准确性和可靠性。以上代码示例展示了如何设置 5 分钟的 Watermark，您可以根据需要修改 maxOutOfOrderness 参数来调整 Watermark 的延迟。