前两天讲了 Flink 的基础和 DataStream API。但无界流上的聚合（sum/reduce）会无限累积状态，实际业务通常需要"按时间段统计"，比如"每分钟的点击量"。这就是窗口（Window）的作用。本篇讲 Flink 的时间语义、水位线、四种窗口类型，并重点讲透滚动窗口的原理和使用。
一、为什么需要窗口
无界流上的聚合有两个问题：
- 状态无限增长：
keyBy().sum()会保存每个 key 的所有历史值，内存会爆； - 无法按时间段统计：业务通常要"过去 1 分钟"、"过去 1 小时"的统计，而不是从开始到现在。
窗口的作用：把无界流切成有限的、按时间或数量划分的"桶"，在每个桶内做聚合。
二、时间语义
窗口基于时间划分，所以先要搞清楚 Flink 的三种时间语义。
2.1 三种时间
数据产生 ──────► 进入Flink ──────► 被算子处理
│ │ │
事件时间 摄入时间 处理时间
(Event Time) (Ingestion Time) (Processing Time)
| 时间 | 定义 | 特点 | 适用场景 |
|---|---|---|---|
| 事件时间 | 数据本身携带的时间戳 | 准确，能处理乱序和迟到 | 生产推荐 |
| 摄入时间 | 数据进入 Source 的时间 | 折中，不能处理 source 内部乱序 | 一般不用 |
| 处理时间 | 算子所在机器的系统时间 | 简单、快，但结果不确定 | 对时间精度要求不高 |
为什么推荐事件时间：数据可能因为网络延迟、Kafka 堆积等原因乱序到达，用事件时间才能保证统计结果的正确性。比如 10:00 产生的数据可能 10:05 才到，如果用处理时间会算到 10:05 的窗口里，结果就错了。
2.2 设置时间语义
// 事件时间（推荐）
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
// 处理时间
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
注意：Flink 1.12+ 默认就是事件时间，不需要显式设置。但需要给数据分配时间戳和水位线。
三、水位线（Watermark）
3.1 为什么需要水位线
事件时间模式下，数据可能乱序到达。比如：
数据顺序：10:01 的数据先到，10:00 的数据后到
Flink 怎么知道"10:00 的数据都到齐了，可以计算 10:00 这个窗口了"？答案是水位线。
3.2 水位线的定义
水位线是一个特殊的时间戳，代表"小于这个时间戳的数据都已经到齐了"。
比如水位线推进到 10:00:05，就告诉 Flink：10:00:05 之前的数据应该都到了，如果还有晚于这个时间到达但时间戳小于 10:00:05 的数据，就是"迟到数据"。
3.3 水位线的生成
水位线 = 当前看到的最大事件时间 - 容忍的乱序时间。
比如最大乱序是 5 秒，当前看到的最大时间戳是 10:00:10，那么水位线 = 10:00:10 - 5s = 10:00:05。
// 基于乱序时间的水位线生成（最常用）
WatermarkStrategy<Event> strategy = WatermarkStrategy
.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5)) // 最大乱序 5 秒
.withTimestampAssigner((event, timestamp) -> event.getTs()); // 从数据提取时间戳
DataStream<Event> withWatermark = events.assignTimestampsAndWatermarks(strategy);
3.4 水位线的传递
- 每个 Task 独立生成水位线；
- 上游 Task 把水位线广播给所有下游 Task；
- 下游 Task 取所有上游水位线的最小值作为自己的水位线；
- 当水位线超过窗口结束时间时，触发窗口计算。
多并行度下的水位线：取所有上游的最小值，保证最慢的那个分区的数据也到齐了。
四、窗口类型
Flink 提供四种窗口：
| 窗口类型 | 说明 | 适用场景 |
|---|---|---|
| 滚动窗口（Tumbling） | 窗口大小固定，不重叠 | 每分钟、每小时统计 |
| 滑动窗口（Sliding） | 窗口大小固定，有重叠 | 每 5 秒统计过去 1 分钟 |
| 会话窗口（Session） | 按活动间隔划分 | 用户会话分析 |
| 全局窗口（Global） | 所有数据一个窗口，需自定义触发器 | 特殊场景 |
4.1 窗口的组成
stream
.keyBy(...) // 按键分区（必须）
.window(...) // 窗口分配器（决定窗口类型）
.trigger(...) // 触发器（可选，何时触发计算）
.evictor(...) // 移除器（可选，计算前后移除数据）
.apply/reduce/aggregate // 窗口函数（计算逻辑）
五、滚动窗口（重点）
5.1 什么是滚动窗口
滚动窗口的特点：窗口大小固定，窗口之间不重叠，数据只属于一个窗口。
时间线： 0 10 20 30 40 50 60
窗口1： [--------)
窗口2： [--------)
窗口3： [--------)
窗口4： [--------)
上图是大小为 20 的滚动窗口，每个数据只落在一个窗口里。
5.2 什么时候用滚动窗口
- 按固定时间周期统计：每分钟 UV、每小时销售额、每天订单量；
- 不需要重叠统计的场景。
5.3 滚动窗口代码
方式一：用 TumblingProcessingTimeWindows（处理时间）
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
DataStream<Tuple2<String, Integer>> result = stream
.keyBy(t -> t.f0)
.window(TumblingProcessingTimeWindows.of(Time.minutes(1))) // 1 分钟滚动窗口
.sum(1);
方式二：用 TumblingEventTimeWindows（事件时间，推荐）
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
DataStream<Tuple2<String, Integer>> result = stream
.keyBy(t -> t.f0)
.window(TumblingEventTimeWindows.of(Time.minutes(1))) // 1 分钟滚动窗口
.sum(1);
方式三：快捷方法（Flink 1.13+ 推荐）
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
// 处理时间
result = stream.keyBy(t -> t.f0)
.window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
.sum(1);
// 事件时间
result = stream.keyBy(t -> t.f0)
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.sum(1);
5.4 滚动窗口的对齐
滚动窗口默认和**时间戳 0（1970-01-01 00:00:00 UTC）**对齐。比如 1 分钟滚动窗口，窗口边界是 00:00、00:01、00:02…
如果需要偏移对齐（比如窗口从 00:00:30 开始），可以加 offset：
// 窗口大小 1 分钟，偏移 30 秒
TumblingEventTimeWindows.of(Time.minutes(1), Time.seconds(30))
效果：窗口边界变成 00:00:30、00:01:30、00:02:30…
5.5 滚动窗口的触发时机
- 处理时间：当机器时间超过窗口结束时间，触发计算；
- 事件时间：当水位线超过窗口结束时间，触发计算。
举例：事件时间、1 分钟滚动窗口、最大乱序 5 秒。
数据时间戳到达顺序：
10:00:03, 10:00:58, 10:01:02, 10:00:55（迟到）, 10:01:10
水位线变化：
收到 10:00:03 → 水位线 = 10:00:03 - 5s = 09:59:58
收到 10:00:58 → 水位线 = 10:00:58 - 5s = 10:00:53
收到 10:01:02 → 水位线 = 10:01:02 - 5s = 10:00:57 < 10:01:00，不触发
收到 10:01:10 → 水位线 = 10:01:10 - 5s = 10:01:05 ≥ 10:01:00，触发 [10:00, 10:01) 窗口计算
窗口 [10:00, 10:01) 计算时包含：10:00:03、10:00:58、10:00:55（虽然 10:00:55 晚到，但水位线还没到 10:01:00，所以还在窗口内）。
10:01:10 之后到达的、时间戳 < 10:01:00 的数据就是迟到数据。
六、其他窗口类型
6.1 滑动窗口
窗口大小固定，滑动步长固定，窗口之间可能重叠。
窗口大小 10，步长 5：
时间：0 5 10 15 20 25
窗口1：[---------)
窗口2： [---------)
窗口3： [---------)
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
// 窗口大小 1 分钟，滑动步长 10 秒
.stream.keyBy(t -> t.f0)
.window(SlidingEventTimeWindows.of(Time.minutes(1), Time.seconds(10)))
.sum(1);
适用场景：需要"过去 N 分钟"的实时统计，比如每 10 秒更新一次过去 1 分钟的平均延迟。
6.2 会话窗口
按数据的间隔划分窗口，两次数据间隔超过阈值就开新窗口。
数据：A B C D E F
间隔： <5s <5s >10s <5s <5s >10s
窗口： [ A B C ] [ D E ] [ F ]
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
// 会话间隔 10 分钟
.stream.keyBy(t -> t.f0)
.window(EventTimeSessionWindows.withGap(Time.minutes(10)))
.sum(1);
适用场景：用户会话分析，用户连续操作算一个会话，超过 10 分钟没操作算会话结束。
6.3 全局窗口
所有数据放到一个窗口，需要自定义触发器决定何时计算。
stream.keyBy(t -> t.f0)
.window(GlobalWindows.create())
.trigger(CountTrigger.of(100)) // 每 100 条数据触发一次
.sum(1);
一般不用，除非有特殊需求。
七、窗口函数
窗口分配器决定数据进哪个窗口，窗口函数决定怎么计算。
7.1 增量聚合函数
每来一条数据就更新中间状态，窗口触发时输出结果。内存占用小，速度快。
| 函数 | 说明 |
|---|---|
reduce | 同类型输入输出，自定义聚合 |
aggregate | 输入输出类型可以不同，更灵活 |
sum/min/max | 内置聚合 |
// aggregate 示例：求窗口内平均值
stream.keyBy(t -> t.f0)
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.aggregate(new AggregateFunction<Tuple2<String, Integer>, Tuple2<Integer, Integer>, Double>() {
// 累加器：(count, sum)
@Override
public Tuple2<Integer, Integer> createAccumulator() {
return Tuple2.of(0, 0);
}
@Override
public Tuple2<Integer, Integer> add(Tuple2<String, Integer> value, Tuple2<Integer, Integer> acc) {
return Tuple2.of(acc.f0 + 1, acc.f1 + value.f1);
}
@Override
public Double getResult(Tuple2<Integer, Integer> acc) {
return acc.f0 == 0 ? 0.0 : acc.f1.doubleValue() / acc.f0;
}
@Override
public Tuple2<Integer, Integer> merge(Tuple2<Integer, Integer> a, Tuple2<Integer, Integer> b) {
return Tuple2.of(a.f0 + b.f0, a.f1 + b.f1);
}
});
7.2 全窗口函数
窗口触发时一次性拿到窗口内所有数据，再做计算。内存占用大，但能访问全部数据。
| 函数 | 说明 |
|---|---|
apply | 旧 API，传入 WindowFunction |
process | 新 API，传入 ProcessWindowFunction，能访问窗口元信息 |
// process 示例：求窗口内 TopN
stream.keyBy(t -> t.f0)
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.process(new ProcessWindowFunction<Tuple2<String, Integer>, String, String, TimeWindow>() {
@Override
public void process(String key, Context ctx, Iterable<Tuple2<String, Integer>> values, Collector<String> out) {
List<Tuple2<String, Integer>> list = new ArrayList<>();
values.forEach(list::add);
list.sort((a, b) -> b.f1 - a.f1); // 降序
long windowStart = ctx.window().getStart();
long windowEnd = ctx.window().getEnd();
out.collect(String.format("窗口[%d,%d) %s 的 Top3: %s",
windowStart, windowEnd, key,
list.stream().limit(3).map(t -> t.f1).collect(Collectors.toList())));
}
});
7.3 增量 + 全窗口结合
增量函数计算中间结果，全窗口函数拿到中间结果再输出，兼顾性能和灵活性。
stream.keyBy(t -> t.f0)
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.aggregate(
new MyAggregateFunction(), // 增量聚合
new MyProcessWindowFunction() // 全窗口输出
);
八、迟到数据处理
事件时间模式下，水位线超过窗口结束时间后到达的数据就是迟到数据。
8.1 默认行为
迟到数据被直接丢弃。
8.2 允许迟到（Allowed Lateness）
设置一个迟到容忍时间，窗口计算后不立即销毁，等迟到时间内的数据还能更新结果。
stream.keyBy(t -> t.f0)
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.allowedLateness(Time.minutes(1)) // 允许迟到 1 分钟
.sum(1);
注意：允许迟到期间，窗口会为每个迟到数据重新计算并输出，可能产生多条结果。
8.3 侧边输出迟到数据
把迟到数据收集到旁路流，单独处理（比如写日志或存数据库）。
OutputTag<Tuple2<String, Integer>> lateTag = new OutputTag<>("late"){};
SingleOutputStreamOperator<Tuple2<String, Integer>> result = stream
.keyBy(t -> t.f0)
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.sideOutputLateData(lateTag) // 迟到数据输出到旁路
.sum(1);
DataStream<Tuple2<String, Integer>> lateStream = result.getSideOutput(lateTag);
8.4 三种处理方式对比
| 方式 | 效果 | 适用场景 |
|---|---|---|
| 默认丢弃 | 迟到数据直接扔 | 对完整性要求不高 |
| allowedLateness | 迟到数据更新窗口结果 | 需要结果完整，但迟到不多 |
| sideOutputLateData | 迟到数据单独处理 | 对迟到数据做补偿处理 |
九、完整案例：每分钟点击量统计
9.1 需求
从 Kafka 读取用户点击流，用事件时间 1 分钟滚动窗口统计每个 URL 的点击量。
9.2 代码
public class UrlClickCountPerMinute {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 1. Source：Kafka
KafkaSource<String> source = KafkaSource.<String>builder()
.setBootstrapServers("localhost:9092")
.setTopics("click-log")
.setGroupId("flink-window-demo")
.setValueOnlyDeserializer(new SimpleStringSchema())
.build();
DataStream<String> raw = env.fromSource(source, WatermarkStrategy.noWatermarks(), "kafka");
// 2. 解析并提取事件时间 + 水位线
WatermarkStrategy<Click> strategy = WatermarkStrategy
.<Click>forBoundedOutOfOrderness(Duration.ofSeconds(5))
.withTimestampAssigner((click, ts) -> click.getTs());
DataStream<Click> clicks = raw
.map(line -> {
JSONObject json = JSON.parseObject(line);
return new Click(json.getString("url"), json.getLong("ts"));
})
.assignTimestampsAndWatermarks(strategy);
// 3. 按 URL 分组，1 分钟滚动窗口，统计点击量
DataStream<Tuple2<String, Integer>> result = clicks
.keyBy(c -> c.url)
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.sum(1) // 这里假设 Click 已经转成 (url, 1)
;
// 上面的 sum 需要 Click 是 Tuple，改一下：
DataStream<Tuple2<String, Integer>> result2 = clicks
.map(c -> Tuple2.of(c.url, 1))
.keyBy(t -> t.f0)
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.sum(1);
// 4. 输出
result2.print();
env.execute("URL Click Count Per Minute");
}
public static class Click {
public String url;
public long ts;
public Click() {}
public Click(String url, long ts) { this.url = url; this.ts = ts; }
public long getTs() { return ts; }
}
}
9.3 执行流程
Kafka 数据 → 解析 → 分配时间戳和水位线
→ keyBy(url) → 1分钟滚动窗口
→ 水位线超过窗口结束时间 → 触发 sum 计算
→ 输出每个 URL 在该分钟的点击量
十、典型坑点
坑点 1：窗口不触发计算
现象：窗口一直没有输出。
原因：事件时间模式下没有生成水位线，或水位线推进太慢。
对策：检查 assignTimestampsAndWatermarks 是否设置；确认数据的时间戳正确；处理时间模式用 TumblingProcessingTimeWindows。
坑点 2：水位线一直不前进
现象：某个分区没数据，水位线卡在最小值。
对策：用 withIdleness(Duration.ofMinutes(1)) 标记空闲分区，让它不阻塞水位线。
WatermarkStrategy.<Click>forBoundedOutOfOrderness(Duration.ofSeconds(5))
.withIdleness(Duration.ofMinutes(1))
.withTimestampAssigner(...);
坑点 3：数据倾斜导致某个窗口特别大
某个 key 的数据量特别大，窗口计算时内存爆炸。
对策：拆分大 key（加随机前缀），或用增量聚合函数减少内存占用。
坑点 4：允许迟到后结果重复输出
allowedLateness 会让窗口在迟到数据到达时重新计算并输出。
对策：下游用 upsert 模式（比如写入 Doris/Redis 用主键覆盖），或用 process 手动去重。
坑点 5：事件时间和处理时间搞混
用错窗口分配器（事件时间数据用了 ProcessingTimeWindows），结果不对。
对策：事件时间数据用 TumblingEventTimeWindows，处理时间数据用 TumblingProcessingTimeWindows。
坑点 6：乱序时间设置太大导致延迟高
forBoundedOutOfOrderness 设太大，窗口触发晚，结果延迟高。
对策：根据实际数据乱序情况设置，通常 1-10 秒；太大不如用 allowedLateness。
十一、三天知识串联
| 天 | 主题 | 核心 |
|---|---|---|
| Day1 | 入门与环境搭建 | Flink 概念、架构、安装、WordCount |
| Day2 | DataStream API | Source → Transformation → Sink、并行度、分区 |
| Day3 | 窗口与时间语义 | 事件时间、水位线、滚动/滑动/会话窗口、迟到数据 |
核心思路：
数据进来（Source）
→ 清洗转换（map/flatMap/filter）
→ 按键分区（keyBy）
→ 按窗口切分（window，推荐滚动窗口）
→ 窗口内聚合（sum/reduce/aggregate）
→ 输出（Sink）
掌握这三天的内容，就能完成大部分实时统计类需求。进阶可以学状态管理（State TTL）、Checkpoint 调优、Flink SQL、CEP 复杂事件处理等。
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/YEEMOLY/article/details/170293648
