java

4.3 Flink 窗口：滚动、滑动、会话、全局

Flink 窗口类型深度解析：TumblingEventTimeWindows 滚动窗口、SlidingEventTimeWindows 滑动窗口、EventTimeSessionWindows 会话窗口、GlobalWindows 自定义触发器、WindowAssigner/Trigger/Evictor 的联合工作机制，以及增量聚合与全量聚合的选择。

Leeting Yan 2026-11-01 3 分钟阅读 1324 字

Flink 窗口模型

Flink 窗口由三部分组成：

WindowAssigner: 给事件分配窗口（滚动/滑动/会话/全局）
    ↓
Trigger: 决定何时触发窗口计算
    ↓
Evictor: 可选，触发前过滤窗口内元素
    ↓
WindowFunction: 计算逻辑（增量或全量）

滚动窗口（Tumbling Window）

窗口大小固定，不重叠。

stream.keyBy(Event::getSymbol)
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
    .aggregate(new CountAggregate());

时间：|----1min----|----1min----|----1min----|----1min----|
事件：    A B           C D E         F              G H I
窗口：   [A,B]         [C,D,E]       [F]            [G,H,I]

适用：每分钟/每小时固定周期的统计。

eventd v3 使用 1 分钟滚动窗口统计每分钟订单量。

滑动窗口（Sliding Window）

窗口大小 > 滑动步长，窗口之间有重叠。

stream.keyBy(Event::getSymbol)
    .window(SlidingEventTimeWindows.of(Duration.ofMinutes(5), Duration.ofMinutes(1)))
    .aggregate(new AverageAggregate());

窗口大小 5min，步长 1min：
[-----5min-----]
[-----5min-----]
[-----5min-----]
...

适用：移动平均（近 5 分钟平均 QPS）、平滑曲线。

会话窗口（Session Window）

按活动间隙动态划分，没有固定大小。

stream.keyBy(Event::getUserId)
    .window(EventTimeSessionWindows.withGap(Duration.ofMinutes(30)))
    .aggregate(new SessionAggregate());

事件：A--B--------C--D--E-----------F--G
间隙：    ↑<30min   ↑30min↑<30min   ↑>30min↑<30min
窗口：   [A,B]      [C,D,E]         [F,G]

适用：用户行为分析（一次浏览会话内的点击序列）。

会话窗口的特殊性：

窗口在数据到达时创建，不是预先划分好的
新数据如果落在两个会话的间隙里，会把两个尚未清理的窗口合并成一个
窗口触发后如果还没被清理，allowedLateness 内的迟到数据仍可能参与合并并再次触发；超出后只能进侧输出

全局窗口（Global Window）

只有一个窗口，包含所有数据。默认不触发，需自定义 Trigger。

stream.keyBy(Event::getSymbol)
    .window(GlobalWindows.create())
    .trigger(CountTrigger.of(100))  // 100 条触发一次
    .aggregate(new BatchAggregate());

适用：批量处理（每 1000 条批量写入数据库）、自定义触发逻辑。

WindowFunction 两种模式

增量聚合（AggregateFunction）

只保存中间状态，内存占用小：

public class CountAggregate implements AggregateFunction<Event, Long, Long> {
    @Override
    public Long createAccumulator() { return 0L; }

    @Override
    public Long add(Event value, Long accumulator) { return accumulator + 1; }

    @Override
    public Long getResult(Long accumulator) { return accumulator; }

    @Override
    public Long merge(Long a, Long b) { return a + b; }
}

stream.keyBy(Event::getSymbol)
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
    .aggregate(new CountAggregate());

全量聚合（WindowFunction）

访问窗口内所有元素：

stream.keyBy(Event::getSymbol)
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
    .apply(new WindowFunction<Event, Result, String, TimeWindow>() {
        @Override
        public void apply(String key, TimeWindow window,
                          Iterable<Event> inputs, Collector<Result> out) {
            List<Event> list = new ArrayList<>();
            inputs.forEach(list::add);
            out.collect(new Result(key, window.getStart(), window.getEnd(), list.size()));
        }
    });

模式	内存	适用	延迟
增量	O(1)	count/sum/min/max	低
全量	O(N)	排序/去重/自定义	高（等窗口满）

自定义 Trigger

public class MetricTrigger extends Trigger<Event, TimeWindow> {
    @Override
    public TriggerResult onElement(Event event, long timestamp,
                                   TimeWindow window, TriggerContext ctx) {
        // 每条事件触发一次
        ctx.registerEventTimeTimer(window.maxTimestamp());
        return TriggerResult.FIRE;  // 立即触发
    }

    @Override
    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) {
        return time == window.maxTimestamp() ? TriggerResult.FIRE_AND_PURGE
                                             : TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) {
        return TriggerResult.CONTINUE;
    }

    @Override
    public void clear(TimeWindow window, TriggerContext ctx) {
        ctx.deleteEventTimeTimer(window.maxTimestamp());
    }
}

TriggerResult：

CONTINUE：继续等待
FIRE：触发计算，但不清理窗口
FIRE_AND_PURGE：触发并清理窗口
PURGE：只清理，不触发

自定义 Evictor

stream.keyBy(Event::getSymbol)
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(5)))
    .evictor(new Evictor<Event, TimeWindow>() {
        @Override
        public void evictBefore(Iterable<TimestampedValue<Event>> elements,
                                int size, TimeWindow window, EvictorContext ctx) {
            // 触发前驱逐
            for (Iterator<TimestampedValue<Event>> it = elements.iterator(); it.hasNext();) {
                if (it.next().getTimestamp() < window.getEnd() - 60000) {
                    it.remove();  // 驱逐窗口最后一分钟之前的数据
                }
            }
        }

        @Override
        public void evictAfter(Iterable<TimestampedValue<Event>> elements,
                               int size, TimeWindow window, EvictorContext ctx) {
            // 触发后驱逐（通常不用）
        }
    });

窗口迟到处理

OutputTag<Event> lateTag = new OutputTag<Event>("late"){};

stream.keyBy(Event::getSymbol)
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
    .allowedLateness(Duration.ofSeconds(10))
    .sideOutputLateData(lateTag)
    .aggregate(new CountAggregate());

eventd v3 窗口配置

eventd v3 使用多种窗口：

// 实时 QPS：1 分钟滚动窗口
stream.keyBy(Event::getSymbol)
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
    .aggregate(new OrderCountAggregate());

// 平均延迟：5 分钟滑动窗口，步长 1 分钟
stream.keyBy(Event::getSymbol)
    .window(SlidingEventTimeWindows.of(Duration.ofMinutes(5), Duration.ofMinutes(1)))
    .aggregate(new AverageLatencyAggregate());

// 用户会话：30 分钟会话窗口
stream.keyBy(Event::getUserId)
    .window(EventTimeSessionWindows.withGap(Duration.ofMinutes(30)))
    .aggregate(new SessionDurationAggregate());

小结

窗口类型	特点	适用场景
滚动	固定大小，不重叠	固定周期统计
滑动	固定大小，有重叠	移动平均
会话	动态大小，按间隙	用户行为
全局	一个窗口	自定义触发

聚合模式	内存	适用
增量	O(1)	count/sum/avg/min/max
全量	O(N)	排序/去重/自定义

第 4 章结束。下一节 5.1 Flink 状态管理：KeyedState 。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页