Windows and timers

The source notes for this track are currently maintained in Chinese.

预计阅读时间: 60 分钟 前置阅读: doc-02(流数据链路) 下一次阅读: doc-06(Watermark)

1. Window 分类与内部表示

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

5. Timer 的性能陷阱

大量 Timer（数百万级）的危害：

1. Heap Timer → 堆膨胀, GC 压力。每次 Watermark 推进 → 遍历 Timer 堆 → O(N log N)
2. RocksDB Timer → 每次 Timer 触发触发 RocksDB I/O → 延迟增加
3. 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