大数据 -122 - Flink Watermark 全面解析：事件时间窗口、乱序处理与迟到数据完整指南

2025-10-12
山东
本文字数：4382 字
阅读完需：约 14 分钟

点一下关注吧！！！非常感谢！！持续更新！！！

🚀 AI 篇持续更新中！（长期更新）

AI 炼丹日志-31- 千呼万唤始出来 GPT-5 发布！“快的模型 + 深度思考模型 + 实时路由”，持续打造实用 AI 工具指南！📐🤖

💻 Java 篇正式开启！（300 篇）

目前 2025 年 10 月 07 日更新到：Java-141 深入浅出 MySQL Spring 事务失效的常见场景与解决方案详解（3）MyBatis 已完结，Spring 已完结，Nginx 已完结，Tomcat 已完结，分布式服务正在更新！深入浅出助你打牢基础！

📊 大数据板块已完成多项干货更新（300 篇）：

包括 Hadoop、Hive、Kafka、Flink、ClickHouse、Elasticsearch 等二十余项核心组件，覆盖离线+实时数仓全栈！大数据-278 Spark MLib - 基础介绍机器学习算法梯度提升树 GBDT 案例详解

章节内容

上节我们完成了如下的内容：

Flink Time 详解
示例内容分析
Watermark

Watermark

Watermark 在窗口计算中的作用

在使用基于事件时间的窗口时，Flink 依赖 Watermark 来决定何时触发窗口计算。Watermark 机制是 Flink 处理乱序事件的核心组件，它本质上是一个特殊的时间戳，表示系统认为在该时间点之前的事件应该都已经到达了。例如：

对于每 10 秒的滚动窗口（例如[00:00-00:10)、[00:10-00:20)等）：
当 Watermark 时间戳达到或超过 00:10 时
Flink 会认为 00:00-00:10 这个窗口的所有事件都已到达（允许一定的延迟）
此时触发该窗口的计算

Watermark 的工作机制包含几个关键点：

乱序处理：允许事件延迟到达，Watermark 会根据最大延迟设置来等待可能的延迟事件
触发条件：只有当 Watermark ≥ 窗口结束时间时才会触发计算
延迟容忍：通过设置适当的 Watermark 间隔和最大延迟时间来平衡计算延迟和结果准确性

典型的应用场景包括：

物联网传感器数据收集（设备可能有网络延迟）
用户行为日志分析（不同地区的用户数据到达时间不一致）
金融交易监控（需要处理网络延迟导致的乱序交易记录）

示例配置：

DataStream<T> stream = ...;stream.assignTimestampsAndWatermarks(    WatermarkStrategy        .<T>forBoundedOutOfOrderness(Duration.ofSeconds(5))        .withTimestampAssigner(...));

复制代码

这个配置表示允许最多 5 秒的延迟，5 秒后仍然没有到达的事件将被视为迟到的数据。

假设有一个 10 秒的窗口，并且 Watermark 达到 12:00:10，此时 Flink 会触发 12:00:00 - 12:00:10 的窗口计算。

如何处理迟到事件

尽管 Watermark 能有效解决乱序问题，但总有可能会出现事件在生成 Watermark 之后才到达的情况（即“迟到事件”）。为此，Flink 提供了处理迟到事件的机制：

允许一定的延迟处理：可以配置窗口允许迟到的时间。
迟到事件的侧输出流（Side Output）：可以将迟到的事件发送到一个侧输出流中，以便后续处理。

DataStream<Tuple2<String, Integer>> mainStream =   stream.keyBy(t -> t.f0)        .window(TumblingEventTimeWindows.of(Time.seconds(10)))        .allowedLateness(Time.seconds(5))        .sideOutputLateData(lateOutputTag);

复制代码

代码实现

数据格式

...01,158648957500001,158648957600001,158648957700001,158648957800001,1586489579000

复制代码

编写代码

这段代码实现了：

通过 socket 获取实时流数据。
将流数据映射成带有时间戳的二元组形式。
应用了一个允许 5 秒乱序的水印策略，确保 Flink 可以处理乱序的事件流。
按照事件的 key 进行分组，并在事件时间的基础上进行 5 秒的滚动窗口计算。
最后输出每个窗口内事件的时间范围、窗口开始和结束时间等信息。

其中，这里对流数据进行了按 key（事件的第一个字段）分组，并且使用了滚动窗口（Tumbling Window），窗口长度为 5 秒。在 apply 方法中，你收集窗口中的所有事件，并根据事件时间戳进行排序，然后输出每个窗口的开始和结束时间，以及窗口中最早和最晚事件的时间戳。

SingleOutputStreamOperator<String> res = waterMark    .keyBy(new KeySelector<Tuple2<String, Long>, String>() {        @Override        public String getKey(Tuple2<String, Long> value) throws Exception {            return value.f0;        }    })    .window(TumblingEventTimeWindows.of(Time.seconds(5)))    .apply(new WindowFunction<Tuple2<String, Long>, String, String, TimeWindow>() {        @Override        public void apply(String s, TimeWindow window, Iterable<Tuple2<String, Long>> input, Collector<String> out) throws Exception {            List<Long> list = new ArrayList<>();            for (Tuple2<String, Long> next : input) {                list.add(next.f1);            }            Collections.sort(list);            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");            String result = "key: " + s + ", list.size(): " + list.size() + ", list.get(0): " + sdf.format(list.get(0)) + ", list.get(last): " + sdf.format(list.get(list.size() - 1))                    + ", start: " + sdf.format(window.getStart()) + ", end: " + sdf.format(window.getEnd());            out.collect(result);        }    });

复制代码

水印的策略，定义了一个 Bounded Out-of-Orderness 的水印策略，允许最多 5 秒的事件乱序，在 extractTimestamp 中，提取了事件的时间戳，并打印出每个事件的 key 和对应的事件时间。还维护了一个 currentMaxTimestamp 来记录当前最大的事件时间戳：

WatermarkStrategy<Tuple2<String, Long>> watermarkStrategy = WatermarkStrategy    .<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))    .withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
        Long currentMaxTimestamp = 0L;        final SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
        @Override        public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {            long timestamp = element.f1;            currentMaxTimestamp = Math.max(currentMaxTimestamp, timestamp);            System.out.println("Key:" + element.f0 + ", EventTime: " + element.f1 + ", " + format.format(element.f1));            return element.f1;        }    });

复制代码

完整代码如下所示，代码实现了一个基于事件时间的流处理系统，并通过水印（Watermark）机制来处理乱序事件：

package icu.wzk;public class WatermarkTest01 {
    public static void main(String[] args) throws Exception {        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        env.setParallelism(1);        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);        DataStreamSource<String> data = env.socketTextStream("localhost", 9999);        SingleOutputStreamOperator<Tuple2<String, Long>> mapped = data.map(                new MapFunction<String, Tuple2<String, Long>>() {                    @Override                    public Tuple2<String, Long> map(String value) throws Exception {                        String[] split = value.split(",");                        return new Tuple2<>(split[0], Long.valueOf(split[1]));                    }                }        );
        WatermarkStrategy<Tuple2<String, Long>> watermarkStrategy = WatermarkStrategy                .<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
                    Long currentMaxTimestamp = 0L;                    final SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
                    @Override                    public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {                        long timestamp = element.f1;                        currentMaxTimestamp = Math.max(currentMaxTimestamp, timestamp);                        System.out.println("Key:" + element.f0 + ", EventTime: " + element.f1 + ", " + format.format(element.f1));                        return element.f1;                    }                });
        SingleOutputStreamOperator<Tuple2<String, Long>> waterMark = mapped                .assignTimestampsAndWatermarks(watermarkStrategy);        SingleOutputStreamOperator<String> res = waterMark                .keyBy(new KeySelector<Tuple2<String, Long>, String>() {                    @Override                    public String getKey(Tuple2<String, Long> value) throws Exception {                        return value.f0;                    }                })                .window(TumblingEventTimeWindows.of(Time.seconds(5)))                .apply(new WindowFunction<Tuple2<String, Long>, String, String, TimeWindow>() {                    @Override                    public void apply(String s, TimeWindow window, Iterable<Tuple2<String, Long>> input, Collector<String> out) throws Exception {                        List<Long> list = new ArrayList<>();                        for (Tuple2<String, Long> next : input) {                            list.add(next.f1);                        }                        Collections.sort(list);                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");                        String result = "key: " + s + ", list.size(): " + list.size() + ", list.get(0): " + sdf.format(list.get(0)) + ", list.get(last): " + sdf.format(list.get(list.size() - 1))                                + ", start: " + sdf.format(window.getStart()) + ", end: " + sdf.format(window.getEnd());                        out.collect(result);                    }                });
        res.print();        env.execute();    }}

复制代码

运行代码

传入数据

在控制台中，输入如下的数据：

查看结果

控制台运行结果如下：

发布于: 刚刚阅读数: 3

原文链接:【http://xie.infoq.cn/article/26dc478ac33508f62a703657b】。文章转载请联系作者。

武子康

关注

永远好奇无限进步 2019-04-14 加入

Hi, I'm Zikang，好奇心驱动的探索者 | INTJ / INFJ 我热爱探索一切值得深究的事物。对技术、成长、效率、认知、人生有着持续的好奇心和行动力。坚信「飞轮效应」，相信每一次微小的积累，终将带来深远的改变。

发布

暂无评论

创作场景

大数据 -122 - Flink Watermark 全面解析：事件时间窗口、乱序处理与迟到数据完整指南

点一下关注吧！！！非常感谢！！持续更新！！！

🚀 AI 篇持续更新中！（长期更新）

💻 Java 篇正式开启！（300 篇）

📊 大数据板块已完成多项干货更新（300 篇）：

章节内容

Watermark

Watermark 在窗口计算中的作用

如何处理迟到事件

代码实现

数据格式

编写代码

运行代码

传入数据

查看结果

武子康

评论