3使用 flink 事件时间抽取数据时间戳为时间时间设置 watermarker 为 30 秒。 10 分

使用 Flink 事件时间，需要在代码中指定事件时间字段，并使用 AssignerWithPeriodicWatermarks 接口实现自定义的水印生成器。以下是示例代码：

// 定义事件类
public class Event {
    private long timestamp;
    // 其他字段省略
    // ...
    public long getTimestamp() {
        return timestamp;
    }
}

// 创建数据流
DataStream<Event> stream = env.addSource(new MySource());

// 指定事件时间字段
stream.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Event>() {
    private long currentMaxTimestamp = 0L;
    private final long maxOutOfOrderness = 30000L; // 最大允许乱序时间为30秒
    @Override
    public long extractTimestamp(Event event, long previousElementTimestamp) {
        long timestamp = event.getTimestamp();
        currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
        return timestamp;
    }
    @Override
    public Watermark getCurrentWatermark() {
        return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
    }
});

在上述代码中，extractTimestamp() 方法指定了事件时间字段为 Event 类中的 timestamp 字段，并将当前最大事件时间保存在 currentMaxTimestamp 变量中。getCurrentWatermark() 方法则根据当前最大事件时间和最大允许乱序时间计算出水印，并返回。这里的水印生成器采用了周期性生成的方式，每隔一段时间就会生成一个新的水印