Flink如何设置时间戳和递增水位线

Flink 可以通过实现 AssignerWithPeriodicWatermarks 或 AssignerWithPunctuatedWatermarks 接口来设置时间戳和递增水位线。

AssignerWithPeriodicWatermarks 接口允许您以固定的时间间隔生成水位线。您需要实现 getCurrentWatermark() 方法来返回当前的水位线，该方法将在固定的时间间隔内调用。此外，您还需要实现 extractTimestamp() 方法来从事件中提取时间戳。

以下是一个示例：

public class PeriodicAssigner implements AssignerWithPeriodicWatermarks<MyEvent> {

    private final long maxOutOfOrderness = 3500; // 事件最大延迟时间

    private long currentMaxTimestamp;

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

    @Nullable
    @Override
    public Watermark getCurrentWatermark() {
        return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
    }
}

AssignerWithPunctuatedWatermarks 接口允许您根据一些特殊事件（例如，标记事件）生成水位线。您需要实现 checkAndGetNextWatermark() 方法来返回下一个水位线，该方法将在每个事件上调用。

以下是一个示例：

public class PunctuatedAssigner implements AssignerWithPunctuatedWatermarks<MyEvent> {

    private final long maxOutOfOrderness = 3500; // 事件最大延迟时间

    private long currentMaxTimestamp;

    @Nullable
    @Override
    public Watermark checkAndGetNextWatermark(MyEvent lastElement, long extractedTimestamp) {
        if (lastElement.isMarked()) { // 如果事件被标记
            return new Watermark(extractedTimestamp - maxOutOfOrderness);
        }
        return null;
    }

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

在将时间戳和水位线应用于 Flink 流处理应用程序时，您需要使用 assignTimestampsAndWatermarks() 方法将它们分配给数据流源。例如：

DataStream<MyEvent> events = ...;

events.assignTimestampsAndWatermarks(new PeriodicAssigner());

或者：

DataStream<MyEvent> events = ...;

events.assignTimestampsAndWatermarks(new PunctuatedAssigner());
``