1. Window 分类与内部表示
| 窗口 | 行为 | 状态大小 | 典型场景 |
|---|---|---|---|
| Tumbling | 固定大小, 不重叠 | 窗口内数据 | 每分钟 PV |
| Sliding | 固定大小, 可重叠 | 窗口重叠部分共享 | 过去5min, 每1min更新 |
| Session | 动态大小, 按 gap 分割 | 不活跃期间累积 | 用户行为 Session |
| Global | 全量 (单窗口) | 持续增长 | 需自定义 Trigger |
2. WindowAssigner → Trigger → Evictor 链条
// WindowOperator 中 Assigner-Trigger-Evictor 的协作
stream
.keyBy(e -> e.userId)
.window(TumblingEventTimeWindows.of(Time.minutes(1))) // WindowAssigner
.trigger(EventTimeTrigger.create()) // Trigger
.evictor(CountEvictor.of(10)) // Evictor (可选)
.process(new MyWindowFunction()) // WindowFunction
WindowAssigner：assignWindows(element, timestamp) 返回该元素所属的窗口集合（Tumbling=1个, Sliding=多个）
Trigger：决定何时 FIRE/FIRE_AND_PURGE/CONTINUE
Evictor：窗口触发前/后驱逐不需要的元素（如保留最近 10 条）
WindowFunction：窗口触发时的计算逻辑
3. Timer 服务 — 窗口的定时引擎
3.1 Timer 的两种时间域
EventTime Timer: Watermark ≥ timer.timestamp → onEventTime() 触发
ProcessingTime Timer: System.currentTimeMillis() ≥ timer.timestamp → onProcessingTime() 触发
3.2 Timer 的存储结构
// InternalTimer 结构体
class InternalTimer<K, N> {
K key; // Timer 关联的 Key
N namespace; // Timer 的命名空间 (通常 = 窗口)
long timestamp; // 触发时间
}
// Timer 存储:
// Heap 模式: PriorityQueue<InternalTimer>, 按 timestamp 排序
// RocksDB 模式: RocksDB 的 PriorityQueue 实现 (基于 RocksDB 的 KV 存储)
// - Key: (keyGroup, namespace, timestamp, key) 的组合编码
// - Value: 空 (不需要, 只存 Key)
3.3 Timer 的源码流程
// WindowOperator.java — 元素处理
public void processElement(StreamRecord<IN> element) throws Exception {
// 1. WindowAssigner: 计算元素所属的窗口集合
Collection<W> elementWindows = windowAssigner.assignWindows(
element.getValue(), element.getTimestamp(), windowAssignerContext);
// 2. 每个窗口: 存储元素
for (W window : elementWindows) {
windowState.add(element.getValue()); // 存入 ListState
// 3. 注册 Timer (窗口结束时间)
long triggerTime = window.maxTimestamp(); // Sliding: 多个窗口的多个时间
triggerContext.registerEventTimeTimer(triggerTime);
// 4. Trigger: 检查是否该立即触发
TriggerResult result = trigger.onElement(element.getValue(),
element.getTimestamp(), window, triggerContext);
if (result.isFire()) {
fire(window, windowState); // 执行窗口函数
}
if (result.isPurge()) {
cleanup(window); // 清除窗口状态 + Timer
}
}
}
// WindowOperator.java — Timer 触发
public void onEventTime(InternalTimer<K, W> timer) throws Exception {
W window = timer.getNamespace();
// 1. Trigger: 判断是否触发
TriggerResult result = trigger.onEventTime(timer.getTimestamp(), window, triggerContext);
if (result.isFire()) {
fire(window, windowState); // 执行窗口函数
}
if (result.isPurge()) {
cleanup(window); // 清除状态和 Timer → 窗口 lifecycle 结束
}
}
3.4 Timer 的优先级队列实现
// HeapPriorityQueue.java — 堆实现 (O(log n) 插入/删除)
public class HeapPriorityQueue<T extends HeapPriorityQueueElement> {
private T[] queue; // 堆数组
private int size;
private Comparator<T> comparator; // 按 timestamp 排序
public void add(T element) {
queue[++size] = element;
siftUp(size); // heapify up → O(log n)
}
public T peek() {
return size > 0 ? queue[1] : null; // O(1) 获取最小 timestamp 的 Timer
}
public T poll() {
T result = queue[1];
queue[1] = queue[size--];
siftDown(1); // heapify down → O(log n)
return result;
}
}
// RocksDBPriorityQueue.java — RocksDB 存储 (O(log n) 但每次有 I/O 开销)
// 使用 RocksDB 的 KV 存储实现优先级队列
// Key = (keyGroup, namespace, timestamp, key) → 按 RocksDB 的 Key 顺序遍历 = 按 timestamp 遍历
4. 增量 Window vs 全量 Window
| 类型 | 数据结构 | 内存 | 延迟 | 灵活性 |
|---|---|---|---|---|
| ReduceFunction | 单个中间结果 | O(1) | 无 | 低 (只能 reduce) |
| AggregateFunction | 单个累加器 | O(1) | 无 | 中 (输入/累加器/输出类型可不同) |
| ProcessWindowFunction | 所有元素 | O(N) | 需缓存所有元素 | 高 (完全控制) |
| AggregateFunction + ProcessWindowFunction | 累加器 + Process 上下文 | O(1) for state, O(1) for ProcessWindowResult | 无 | 高 |
5. Timer 的性能陷阱
大量 Timer（数百万级）的危害：
- Heap Timer → 堆膨胀, GC 压力。每次 Watermark 推进 → 遍历 Timer 堆 → O(N log N)
- RocksDB Timer → 每次 Timer 触发触发 RocksDB I/O → 延迟增加
- Checkpoint 时 → 序列化所有 Timer → Checkpoint Size 暴增
最佳实践：
- 设置
allowedLateness限制 Timer 存活时间 - 在
clear()中清理 Timer（deleteEventTimeTimer()） - 对于超大状态 → 使用 KeyedProcessFunction 手动管理 Timer（替代 Window）
6. 常见问题 / 面试题
Q1: 增量 Window 和全量 Window 的本质区别？
A: 增量（Reduce/Aggregate）在元素到达时立即更新中间结果，只存储一个累加器（如 count=5, sum=100）。全量（ProcessWindowFunction）缓存所有元素到 ListState，触发时批量处理。增量 → 内存 O(1) 但功能受限；全量 → 内存 O(N) 但完全灵活。
Q2: Timer 过多导致 Checkpoint 慢，如何优化？
A: (1) allowedLateness 限制窗口清理时间；(2) 使用 Set 记录是否需要 Timer → 减少重复注册；(3) 对于极高频率的 Timer（如每 1ms），改为 service 的批处理（ProcessingTimeService 的 scheduleAtFixedRate）。
Q3: Watermark 不触发窗口，但数据在窗口内，是什么原因？
A: (1) Watermark = min(all inputs) → 其中某个 Source 的 Watermark 未前进 → withIdleness()；(2) 事件时间戳错误（未来时间）→ Watermark = maxTimestamp(BoundedOutOfOrderness) 可能被错误的未来时间戳推高 → TimestampsAndWatermarksOperator 有最大时间戳限制。
下一步
- Watermark: doc-06-watermark-and-event-time.md