Flink滑动窗口实战：5分钟统计最近一小时热门商品

1. 实时统计场景的整体设计思路

1.1 为什么选“每隔5分钟统计最近一小时”这个需求

“每隔5分钟统计最近一小时热门商品”这个需求，在实时计算领域里属于非常经典的滑动窗口聚合场景。它看起来简单，但几乎把流处理里最容易踩坑的几个点全踩了一遍：事件时间、水位线、窗口重叠、状态管理、结果输出频率。

先把这个需求翻译成人话：假设用户在电商平台不断点击、加购、下单商品，这些行为会以日志流的形式源源不断进来。我需要每5分钟就输出一次“过去60分钟内，哪些商品最热门”的榜单。注意，这里的“过去一小时”不是自然整点的一小时，而是相对于当前时刻往前滚动的一小时，每5分钟滚动一次。

这就意味着，同一个商品的行为数据，可能会同时落在多个窗口里。比如一件商品在10:03被点击，那么10:05、10:10、10:15……直到11:00之前的每一次统计，它都应该被算进去。这就是滑动窗口（Sliding Window）的典型特征：窗口长度（size）是1小时，滑动步长（slide）是5分钟。

为什么不用滚动窗口（Tumbling Window）？因为滚动窗口是首尾相接、不重叠的，每5分钟只统计这5分钟内的数据，那就不叫“最近一小时”了。所以窗口类型的选择，是由业务语义直接决定的，这一点在动手写代码之前必须先想清楚。

1.2 两种实现路线的取舍：DataStream API 还是 Flink SQL

标题里提到了两条路线：DataStream API 和 Flink SQL。这两条路都能实现同一个需求，但适用场景和上手难度差别很大。

Flink SQL 路线的优势在于表达简洁。一个带滑动窗口的 GROUP BY 查询，几行 SQL 就能把窗口定义、聚合逻辑、时间语义全部说清楚。对于熟悉 SQL 的同学来说，几乎零学习成本，而且 Flink 会自动帮你处理窗口的创建、状态的清理、结果的输出。缺点是灵活性受限，一旦业务逻辑复杂到需要自定义函数、复杂事件处理、多流关联，SQL 就会变得很别扭。

DataStream API 路线的优势在于完全可控。窗口分配器、触发器、驱逐器、允许延迟、侧输出流，每一个环节你都能自己掌控。适合需要精细调优、或者业务逻辑无法用 SQL 表达的场景。缺点是代码量大，而且窗口相关的 API 细节多，稍不注意就会出现结果不对、状态爆炸、数据丢失等问题。

我的建议是：新手先用 SQL 把需求跑通，理解窗口的语义；然后再用 DataStream API 重写一遍，理解底层发生了什么。两条路线对照着看，对滑动窗口的理解会深刻很多。下面我也会按这个顺序来讲。

1.3 数据模型与核心字段设计

不管走哪条路线，输入数据的结构得先定下来。商品行为日志通常包含这几个核心字段：

这里有个关键点：事件时间（Event Time）必须用数据里自带的时间戳，而不是处理时间（Processing Time）。为什么？因为网络延迟、上游积压等原因，数据到达计算引擎的顺序和时间，跟它实际发生的时间往往对不上。如果用处理时间，那么“最近一小时”统计的就是“最近一小时到达的数据”，而不是“最近一小时发生的行为”，结果会有偏差。用事件时间配合水位线，才能保证统计口径正确。

热门商品的“热度”怎么定义？最简单的做法是统计每个商品的点击量（pv），点击越多越热门。如果想更精细，可以给不同行为加权，比如点击算1分、加购算3分、购买算10分，这样得到的榜单更贴近真实转化价值。这个权重设计属于业务层面的取舍，代码里用一个简单的映射就能实现。

2. Flink SQL 实现滑动窗口统计

2.1 建表与时间属性定义

用 SQL 实现的第一步，是把数据流映射成一张带时间属性的表。这里用 DDL 的方式定义，数据源假设是 Kafka（换成其他源也一样，核心是时间属性的声明）。

CREATE TABLE item_behavior ( user_id BIGINT, item_id BIGINT, category_id INT, behavior STRING, ts BIGINT, -- 声明事件时间属性，ts 是毫秒时间戳 event_time AS TO_TIMESTAMP_LTZ(ts, 3), -- 基于事件时间定义水位线，允许5秒乱序 WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND ) WITH ( 'connector' = 'kafka', 'topic' = 'item_behavior', 'properties.bootstrap.servers' = 'localhost:9092', 'properties.group.id' = 'hot_item_group', 'scan.startup.mode' = 'latest-offset', 'format' = 'json' );

这段 DDL 里有几个细节值得展开说。

TO_TIMESTAMP_LTZ(ts, 3)里的3表示时间戳精度是毫秒。如果你的数据是秒级时间戳，这里要改成0。这个参数写错，时间会差出好几个数量级，窗口就完全不对了。

水位线WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND表示：系统认为事件时间小于“当前最大事件时间减5秒”的数据都已经到齐了。这个5秒是乱序容忍度，设得太小，迟到数据会被丢弃；设得太大，窗口触发会延迟，结果输出不及时。实际生产中，这个值要根据上游数据的乱序程度来定，一般先观察数据延迟分布，取P99延迟作为参考。

提示：水位线是 Flink 事件时间机制的基石。它本质上是一个“进度信号”，告诉系统“这个时间点之前的数据我基本都收到了，可以触发窗口计算了”。没有水位线，窗口永远不知道该什么时候关闭和输出。

2.2 滑动窗口的 SQL 写法与参数含义

表建好之后，核心查询就来了。Flink SQL 里滑动窗口用HOP函数（旧版本叫SLIDE）来定义：

CREATE TABLE hot_item_result ( window_end TIMESTAMP(3), item_id BIGINT, pv_count BIGINT ) WITH ( 'connector' = 'print' ); INSERT INTO hot_item_result SELECT HOP_END(event_time, INTERVAL '5' MINUTE, INTERVAL '1' HOUR) AS window_end, item_id, COUNT(*) AS pv_count FROM item_behavior WHERE behavior = 'pv' GROUP BY HOP(event_time, INTERVAL '5' MINUTE, INTERVAL '1' HOUR), item_id;

HOP(event_time, INTERVAL '5' MINUTE, INTERVAL '1' HOUR)这个函数有两个参数，顺序很关键：第一个是滑动步长（slide），第二个是窗口长度（size）。很多人第一次写会记反，结果窗口语义完全错了。记住口诀：先步长后长度，小的在前大的在后。

这个 HOP 会把每条数据分配到多个窗口里。具体来说，一条发生在 10:03 的数据，会被分配到窗口 [09:05, 10:05)、[09:10, 10:10)、[09:15, 10:15)……一直到 [10:00, 11:00)。总共 12 个窗口（60分钟除以5分钟）。这就是滑动窗口“一条数据进多个窗口”的本质，也是它比滚动窗口更耗资源的原因。

HOP_END返回窗口的结束时间，用来标识这个统计结果属于哪个窗口。也可以用HOP_START或HOP_ROWTIME，看业务需要。输出到结果表后，每5分钟就会有一批新的统计结果产生。

2.3 SQL 路线的性能与状态考量

SQL 写起来简单，但底层的状态开销不能忽视。滑动窗口的窗口数量是size / slide，这里是 12 个。也就是说，每个 key（商品）在任意时刻，最多要维护 12 个窗口的聚合状态。如果商品数量是百万级，那状态规模就是千万级，对内存和 checkpoint 都是压力。

有几个优化方向可以聊一聊。

第一，缩小窗口重叠度。如果业务能接受，把 slide 调大（比如10分钟），窗口数量就减半。但这样输出频率就降低了，需要权衡。

第二，提前过滤。上面的 SQL 里加了WHERE behavior = 'pv'，只统计点击行为。如果不加这个过滤，所有行为都参与聚合，状态会更大。能在 SQL 层过滤掉的，就不要留到后面。

第三，考虑用滚动窗口+外部聚合替代。如果只是要“最近一小时”的近似结果，可以用1小时的滚动窗口，然后每5分钟读一次最新结果。这样状态只有1个窗口，开销小很多，代价是结果不是精确的滑动统计。这是一种常见的工程折中。

注意：Flink SQL 的滑动窗口在状态后端里存储的是增量聚合结果（如果用的是 COUNT、SUM 这类可增量计算的聚合），不是原始数据。所以状态大小跟窗口内数据量无关，只跟 key 数量和窗口数量有关。这一点比很多人想象的要乐观。

3. DataStream API 实现与底层机制

3.1 窗口分配器与增量聚合函数

用 DataStream API 实现，代码结构会清晰很多，但每一步都要自己写。核心流程是：source -> keyBy -> window -> aggregate -> sink。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); env.getConfig().setAutoWatermarkInterval(1000L); DataStream<ItemBehavior> stream = env .addSource(new FlinkKafkaConsumer<>("item_behavior", new JSONDeserializationSchema(), props)) .assignTimestampsAndWatermarks( WatermarkStrategy.<ItemBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, ts) -> event.getTs()) ); DataStream<HotItemResult> result = stream .filter(e -> "pv".equals(e.getBehavior())) .keyBy(ItemBehavior::getItemId) .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5))) .aggregate(new PvCountAggregate(), new HotItemWindowFunction()); result.addSink(new PrintSinkFunction<>()); env.execute("HotItemSlidingWindow");

SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5))这行就是滑动窗口分配器。参数顺序是先窗口长度，后滑动步长，跟 SQL 里的 HOP 正好相反。这个顺序差异坑过不少人，写的时候一定要看清楚。

aggregate方法接收两个参数：增量聚合函数和窗口函数。增量聚合函数每来一条数据就更新一次累加器，窗口函数在窗口触发时把累加器结果加工成输出。这种设计的好处是，窗口内不需要缓存原始数据，只需要维护一个累加器（比如一个 Long 计数器），内存占用极小。

public class PvCountAggregate implements AggregateFunction<ItemBehavior, Long, Long> { @Override public Long createAccumulator() { return 0L; } @Override public Long add(ItemBehavior 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; } }

这个累加器就是一个 Long，简单到极致。对比一下，如果用apply方法配合ProcessWindowFunction全量缓存窗口内所有元素，那内存开销就是天壤之别。所以能用增量聚合就别用全量聚合，这是流处理的一条铁律。

3.2 水位线、允许延迟与侧输出流

窗口什么时候触发计算？答案是：当水位线超过窗口结束时间时触发。比如窗口 [09:05, 10:05)，当水位线推进到 10:05 之后，这个窗口就会触发。

但现实中总有迟到数据。一条本该在 10:04 到达的数据，因为网络抖动，10:07 才到。这时候窗口已经触发过了，这条数据怎么办？默认情况下会被丢弃。为了处理这种情况，Flink 提供了allowedLateness：

.window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5))) .allowedLateness(Time.minutes(2)) .sideOutputLateData(new OutputTag<ItemBehavior>("late-data"){}) .aggregate(new PvCountAggregate(), new HotItemWindowFunction());

allowedLateness(Time.minutes(2))表示窗口触发后，还会再保留2分钟的状态，等待迟到数据。每来一条迟到数据，窗口会重新计算并再次输出结果。这意味着下游会收到同一个窗口的多次输出，需要做去重或幂等处理。

sideOutputLateData把超过允许延迟的数据收集到侧输出流，可以单独处理，比如写入日志分析迟到原因，或者补偿到其他系统。

提示：allowedLateness 会增加状态保留时间，进而增加状态大小。设置的时候要权衡：设得太短，迟到数据丢失；设得太长，状态膨胀。一般建议根据业务对数据准确性的要求来定，2到5分钟是比较常见的值。

3.3 窗口结果的输出与下游消费

窗口触发后，HotItemWindowFunction会把结果发往下游。这里有个容易被忽略的点：滑动窗口每5分钟触发一批，但每批包含的是不同窗口的结果。比如 10:05 触发的是 [09:05, 10:05) 窗口，10:10 触发的是 [09:10, 10:10) 窗口。下游收到的是一个个独立的窗口结果，需要根据窗口结束时间来判断这是哪一批统计。

public class HotItemWindowFunction extends ProcessWindowFunction<Long, HotItemResult, Long, TimeWindow> { @Override public void process(Long itemId, Context ctx, Iterable<Long> elements, Collector<HotItemResult> out) { Long count = elements.iterator().next(); out.collect(new HotItemResult(ctx.window().getEnd(), itemId, count)); } }

输出结果里带上window.getEnd()，下游就能知道这个统计属于哪个时间窗口。如果要生成“Top N 热门商品”榜单，还需要在窗口结果之上再做一次排序。这个排序可以用另一个窗口操作实现，也可以推给下游的存储（比如 Redis 的 Sorted Set）来做。

实际生产中，窗口结果通常会写入消息队列或数据库，再由前端服务读取展示。如果直接 print 到控制台，那只是 demo 级别的验证。写入外部系统时要注意：窗口结果是批量产生的，写入时要考虑批量提交和幂等，避免重复写入导致数据翻倍。

4. 常见问题与排查技巧实录

4.1 窗口不触发、结果不输出的排查思路

这是新手最常遇到的问题：代码跑起来了，数据也进来了，但就是看不到输出。排查方向有这么几个。

第一，检查水位线是否推进。窗口触发依赖水位线，如果水位线一直不前进，窗口永远不会触发。水位线不前进的常见原因是：数据源没有新数据进来，或者时间戳字段解析失败导致水位线无法计算。可以在代码里加一行打印，观察水位线的变化：

.assignTimestampsAndWatermarks( WatermarkStrategy.<ItemBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, ts) -> event.getTs()) ).map(new MapFunction<ItemBehavior, ItemBehavior>() { @Override public ItemBehavior map(ItemBehavior value) { System.out.println("Event time: " + value.getTs() + ", current watermark: " + ((EventTimeExtractor) getRuntimeContext().getTimestampExtractor()).getCurrentWatermark()); return value; } });

第二，检查并行度与空闲分区。如果数据源有多个分区，某个分区长时间没有数据，它的水位线就不会推进，进而拖累整个算子的水位线。Flink 提供了withIdleness来处理空闲分区：

WatermarkStrategy.<ItemBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, ts) -> event.getTs()) .withIdleness(Duration.ofMinutes(1))

withIdleness(Duration.ofMinutes(1))表示某个分区超过1分钟没有数据，就把它标记为空闲，水位线计算时忽略它。这个配置在多分区场景下几乎是必加的。

第三，检查窗口时间单位。Time.hours(1)和Time.minutes(5)别写反，也别把hours写成seconds。这种低级错误排查起来最费时间，因为代码不报错，只是结果不对。

4.2 数据倾斜与状态膨胀的应对

商品行为数据天然是不均匀的，头部商品的数据量可能是长尾商品的成千上万倍。这会导致数据倾斜：某些 key 对应的窗口状态特别大，某些特别小。倾斜严重时，个别算子会成为瓶颈，拖慢整个作业。

应对数据倾斜，有几个实用手段。

加盐打散。对热点 key 加随机前缀，先做局部聚合，再去掉前缀做全局聚合。比如把item_id变成item_id + "_" + random(0, 9)，这样热点商品被分散到10个 key 上，聚合压力就分散了。代价是需要两阶段聚合，代码复杂度上升。

预聚合。在进入窗口之前，先做一层本地聚合，减少进入窗口的数据量。比如在 source 后面加一个keyBy(item_id).count()，把原始行为流压缩成计数流，再进入滑动窗口。这样窗口处理的数据量大幅减少。

状态 TTL。如果某些商品长期没有新数据，它的窗口状态应该被清理掉。Flink 的状态 TTL 可以自动清理过期状态：

StateTtlConfig ttlConfig = StateTtlConfig .newBuilder(Time.hours(2)) .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) .build();

注意：状态 TTL 的过期时间是“最后一次访问后多久过期”，不是“创建后多久过期”。对于滑动窗口，窗口状态会持续被访问，所以 TTL 要设得比窗口长度大一些，否则可能误删还在用的状态。

4.3 结果重复与幂等写入的处理

前面提到，allowedLateness会导致同一个窗口多次输出。如果下游是数据库，重复写入就会产生重复记录。解决办法有几种。

方案一：下游做 upsert。如果结果表有主键（比如窗口结束时间+商品ID），用 upsert 语义写入，重复数据会覆盖而不是追加。Flink SQL 的 JDBC connector 支持这种模式。

方案二：结果里带版本号。每次输出带一个递增的版本号，下游只保留最新版本。这需要在窗口函数里维护一个计数器状态。

方案三：不接受迟到数据。如果业务对准确性要求不高，直接不设allowedLateness，窗口触发后就不再更新。这样结果不会重复，但迟到数据会被丢弃。

选择哪种方案，取决于业务对“准确性”和“实时性”的权衡。金融、计费类场景通常要求准确，宁可重复也要保证数据完整；而榜单、推荐类场景对实时性更敏感，可以接受少量误差。

4.4 常见问题速查表

5. 从 Demo 到生产的几个关键补强

5.1 检查点与状态后端配置

Demo 跑通只是第一步，要上生产，检查点（Checkpoint）必须配好。检查点是 Flink 容错的核心，它定期把算子状态快照到持久化存储，作业失败后可以从最近的检查点恢复。

env.enableCheckpointing(60000); // 每60秒做一次检查点 env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000); env.getCheckpointConfig().setCheckpointTimeout(600000); env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

EXACTLY_ONCE模式保证每条数据只被处理一次，这对统计准确性很重要。setMinPauseBetweenCheckpoints避免检查点过于频繁影响正常处理。RETAIN_ON_CANCELLATION表示作业取消时保留检查点，方便后续恢复。

状态后端的选择也影响性能。数据量小的时候用内存状态后端，数据量大时用 RocksDB 状态后端，把状态放到磁盘上，内存压力小很多，代价是读写性能下降。

5.2 并行度与资源规划

并行度决定了作业的并发处理能力。设置得太小，处理不过来；设置得太大，资源浪费，而且可能因为空闲分区拖累水位线。

一个经验公式：并行度 ≈ 峰值吞吐量 / 单并行度处理能力。单并行度处理能力可以通过压测得到，一般单核每秒能处理几万到几十万条简单聚合。假设峰值是每秒50万条，单并行度能处理10万条，那并行度设5就够了，留点余量设8。

资源方面，每个 TaskManager 的 slot 数量、内存大小都要根据状态规模来定。滑动窗口的状态规模 = key 数量 × 窗口数量 × 单窗口状态大小。粗略估算一下，如果商品有100万个，窗口12个，每个窗口状态100字节，那就是 100万 × 12 × 100 = 1.2GB。这还没算上其他开销，所以内存至少要给到2到4GB。

5.3 监控指标与告警设置

生产作业必须配监控，否则出了问题都不知道。Flink 自带了很多指标，重点关注这几个：

• numRecordsIn / numRecordsOut：输入输出记录数，突然掉零说明数据源断了。
• currentWatermark：当前水位线，长时间不推进说明有问题。
• checkpointDuration：检查点耗时，持续增长说明状态太大或存储太慢。
• numLateRecordsDropped：丢弃的迟到数据数，持续增长说明 allowedLateness 设小了。

这些指标可以接入监控系统，设置阈值告警。比如水位线超过10分钟不推进就告警，迟到数据比例超过1%就告警。

提示：监控指标不是配了就行，还要定期回顾。我见过不少作业，告警配了一堆，但没人看，结果故障发生了才发现。建议每周花十分钟看一眼关键指标的趋势，提前发现隐患。

5.4 一个容易忽略的细节：时区问题

时间戳处理里，时区是个隐形杀手。TO_TIMESTAMP_LTZ返回的是带时区的时间戳，如果服务器时区、Flink 配置时区、数据里的时间戳时区不一致，统计出来的窗口时间就会偏移几个小时。

建议统一用 UTC 时间戳存储和计算，展示的时候再转成本地时区。Flink 的table.local-time-zone配置可以指定时区，默认是 UTC。如果数据里的时间戳是本地时间，要么在写入前转成 UTC，要么在 Flink 里配置对应的时区。这个问题在跨时区部署时尤其明显，值得提前确认。

6. 两条路线的对照与选型建议

6.1 SQL 与 DataStream 的能力边界对比

把两条路线放在一起对比，能更清楚地看到各自的适用场景。

SQL 的短板在于“标准之外”的需求。比如你想根据商品类目的不同，动态调整窗口长度，SQL 就很难表达。再比如你想在窗口触发时调用外部服务做二次过滤，SQL 也做不到。这些场景就得回到 DataStream API。

DataStream 的短板在于开发效率。一个 SQL 几行搞定的需求，DataStream 可能要写上百行代码，而且窗口 API 的细节多，容易出错。所以能用 SQL 解决的，优先用 SQL，把 DataStream 留给真正需要它的场景。

6.2 实际项目中的混合使用模式

真实项目里，两条路线经常是混着用的。一个典型的模式是：用 SQL 做 ETL 和标准聚合，用 DataStream 做复杂事件处理和自定义输出。

比如这个热门商品统计，可以用 SQL 完成窗口聚合，把结果写入一张中间表；然后用 DataStream 读取这张表，做 Top N 排序、结果去重、写入外部存储。这样既享受了 SQL 的开发效率，又保留了 DataStream 的灵活性。

Flink 提供了 SQL 和 DataStream 互相转换的 API。tableEnv.toDataStream(table)可以把 SQL 查询结果转成 DataStream，tableEnv.fromDataStream(stream)可以把 DataStream 注册成表。这种互转能力让混合编程变得很自然。

6.3 给不同阶段读者的学习路径建议

如果你是刚接触 Flink 的新手，建议按这个顺序走：

1. 先用 SQL 把需求跑通。建表、定义窗口、聚合、输出，把整条链路走一遍，理解滑动窗口的语义。
2. 再用 DataStream API 重写。重点理解窗口分配器、水位线、增量聚合函数、窗口函数这几个核心概念。
3. 然后研究底层机制。窗口的触发时机、状态的存储结构、检查点的执行流程，这些决定了你能不能调优。
4. 最后做生产化改造。检查点配置、并行度规划、监控告警、时区处理，这些是 Demo 和生产的差距所在。

如果你已经有流处理经验，可以直接从 DataStream API 入手，重点看窗口的触发器和驱逐器，以及状态 TTL 的配置。这些是滑动窗口调优的关键。

我个人在实际操作中的体会是，滑动窗口这个需求看似简单，但它把流处理的核心概念全串起来了：事件时间、水位线、窗口、状态、容错。把这一个案例吃透，比泛泛地看十篇教程都有用。踩过几次坑之后，你会对“数据什么时候到、窗口什么时候触发、结果什么时候输出”这三个问题有肌肉记忆般的直觉，这才是真正入门了。