Apache Flink技术咨询：如何处理水印滞后事件并导出丢弃事件至S3

嘿，针对你的两个问题，我来一步步给你拆解清楚：

你想要实现的简单水印逻辑，用Flink的AssignerWithPeriodicWatermarks就能轻松搞定。这种方式会周期性生成水印（默认每200ms一次，可配置），正好匹配你的需求。

给你个代码示例，你可以根据自己的事件结构调整：

public class CustomTimestampExtractor implements AssignerWithPeriodicWatermarks<YourEvent> {
    // 固定10秒的延迟
    private final long maxDelayMs = 10000;

    @Override
    public long extractTimestamp(YourEvent event, long previousElementTimestamp) {
        // 这里返回事件自身的时间戳（如果你的事件有这个字段的话）
        // 如果你的场景不需要事件自带的时间，也可以直接返回System.currentTimeMillis()
        return event.getEventTimestamp();
    }

    @Override
    public Watermark getCurrentWatermark() {
        // 水印 = 当前系统时间 - 10秒
        return new Watermark(System.currentTimeMillis() - maxDelayMs);
    }
}

然后把这个时间戳分配器应用到你的数据流上：

DataStream<YourEvent> stream = env.addSource(...)
    .assignTimestampsAndWatermarks(new CustomTimestampExtractor());

关于滚动窗口的触发时间，你的理解完全正确：滚动窗口会在窗口结束时间 + 10秒左右触发（因为水印需要超过窗口结束时间才会触发窗口计算）。

Flink默认会直接丢弃滞后于水印的事件，但我们可以用**侧输出流（Side Output）**把这些迟到事件捞出来，再写入S3。这是生产环境中最常用的方案，步骤也很清晰：

1. 定义侧输出流标签 #

首先给迟到事件打个标签，方便后续获取：

// 泛型要和你的事件类型一致
private static final OutputTag<YourEvent> LATE_EVENTS_TAG = new OutputTag<YourEvent>("late-events") {};

2. 配置窗口算子，将迟到事件发送到侧输出 #

在窗口操作中，通过sideOutputLateData()方法把迟到事件导向侧输出流。如果你想严格只丢弃那些滞后于水印的事件，把allowedLateness设为0即可：

DataStream<WindowResult> windowResult = stream
    .keyBy(event -> event.getYourKey()) // 替换成你的key逻辑
    .window(TumblingEventTimeWindows.of(Time.minutes(1))) // 替换成你的窗口大小
    .allowedLateness(Time.seconds(0)) // 0容忍，超过水印的事件直接进侧输出
    .sideOutputLateData(LATE_EVENTS_TAG)
    .apply(new WindowFunction<YourEvent, WindowResult, String, TimeWindow>() {
        @Override
        public void apply(String key, TimeWindow window, Iterable<YourEvent> input, Collector<WindowResult> out) throws Exception {
            // 这里写你的窗口计算逻辑，比如聚合、统计等
            out.collect(new WindowResult(key, window.getEnd(), input.spliterator().estimateSize()));
        }
    });

如果你想给迟到事件一点“缓冲时间”（比如窗口触发后再等5分钟接收迟到事件），可以把allowedLateness改成Time.seconds(300)，超过这个时间的事件才会进入侧输出。

3. 将侧输出流写入S3 #

Flink提供了内置的文件系统Sink，直接配置就能写入S3：

// 获取侧输出流
DataStream<YourEvent> lateEvents = windowResult.getSideOutput(LATE_EVENTS_TAG);

// 序列化事件并写入S3
lateEvents
    .map(event -> {
        // 把事件序列化为JSON或字符串，方便存储
        ObjectMapper mapper = new ObjectMapper();
        return mapper.writeValueAsString(event);
    })
    .sinkTo(FileSystemSink.forRowFormat(
            new Path("s3://your-bucket-path/late-events/"),
            new SimpleStringSchema()
        )
        .withRollingPolicy(DefaultRollingPolicy.builder()
            .withRolloverInterval(Time.minutes(15)) // 每15分钟生成一个新文件
            .withInactivityInterval(Time.minutes(5)) // 5分钟无数据就滚动文件
            .withMaxPartSize(MemorySize.ofMebiBytes(128)) // 文件最大128MB就滚动
            .build())
        .withBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd/HH")) // 按日期小时分目录
        .build());

这样所有被Flink判定为迟到的事件，就会被自动写入S3的指定路径里了。

内容的提问来源于stack exchange，提问作者balaji