1. 从5分钟窗口统计1小时热门商品说起这个需求到底在算什么电商后台的运营看板里有一类指标几乎天天被盯着——最近一小时的热门商品排行。它不像总销量那样越滚越大也不像实时在线人数那样每秒跳动而是每隔一段时间刷新一次告诉你过去六十分钟里哪些商品被看得最多、加购最多、下单最多。这个需求听起来简单但真要用流式计算把它做扎实里面藏着不少门道。我第一次接触这个场景时脑子里第一反应是不就是个滑动窗口吗。结果真动手才发现滑动窗口的滑动步长、窗口大小、时间语义、状态存储、输出频率每一个参数选错都会让结果要么延迟、要么重复、要么把内存撑爆。这个案例之所以经典是因为它同时踩中了流处理的几个核心考点滚动聚合与滑动窗口的区别、事件时间与处理时间的取舍、增量计算与全量重算的权衡、以及DataStream API和Flink SQL两条技术路线的对比。所谓每隔5分钟统计最近一小时拆开来看是两层时间约束。第一层是统计范围最近一小时也就是窗口长度是60分钟。第二层是触发频率每隔5分钟出一次结果也就是窗口每隔5分钟滑动一次。这两层叠加起来就是一个典型的滑动窗口Sliding Window窗口大小60分钟滑动步长5分钟。每5分钟系统就要对过去60分钟的数据做一次聚合输出一份TopN榜单。这里有个容易被忽略的细节如果窗口大小是60分钟、步长是5分钟那么同一份数据会被多少个窗口覆盖答案是60除以5等于12个窗口。也就是说一条商品点击记录进来之后它会同时属于12个不同的滑动窗口参与12次聚合计算。这个数据放大12倍的特性是滑动窗口最需要警惕的地方也是后面选型和调优的关键出发点。适合读这篇内容的人我大致分三类。第一类是刚学Flink、想找一个完整案例把DataStream API和SQL都跑一遍的入门者第二类是有一定基础、但一直没搞明白滑动窗口和TopN怎么配合的开发者第三类是做实时数仓、需要把这类近N分钟热门指标落地到生产环境的工程师。不管你是哪一类我都会尽量把为什么这么设计讲透而不是只丢一段能跑的代码。提示本文所有代码和配置都基于Flink 1.17及以上版本的常见实践低版本在API细节上可能有差异遇到报错先确认版本。2. 滑动窗口的账要先算清楚为什么是60分钟配5分钟在动手写代码之前我习惯先把窗口的账算明白。很多人上来就写TumblingEventTimeWindows写完发现输出频率不对或者干脆用错了窗口类型。这一节我们把窗口选型的逻辑彻底捋一遍。2.1 滚动窗口、滑动窗口、会话窗口到底差在哪Flink的窗口从大类上分三种滚动窗口Tumbling Window、滑动窗口Sliding Window、会话窗口Session Window。它们的核心区别在于窗口之间是否重叠以及窗口边界是否固定。滚动窗口的窗口之间不重叠首尾相接。比如窗口大小60分钟那么就是0-60、60-120、120-180这样一段段切下去。它的特点是每条数据只属于一个窗口计算量最小但输出频率等于窗口大小——你要么每分钟出一个窗口设1分钟要么每小时出一个窗口设60分钟没法做到范围1小时但5分钟出一次。滑动窗口的窗口之间可以重叠。窗口大小60分钟、步长5分钟那么窗口就是0-60、5-65、10-70这样滑动下去。每条数据属于多个窗口计算量成倍增加但换来的是范围大、频率高的效果正好匹配我们的需求。会话窗口则是按数据活跃度切分一段时间没有数据就关闭窗口适合用户行为会话分析跟固定周期统计这个需求不搭。所以结论很明确每隔5分钟统计最近一小时必须用滑动窗口窗口大小60分钟滑动步长5分钟。用滚动窗口做不到用会话窗口更不对。2.2 一条数据被算12次这个代价能不能接受前面提到60分钟窗口配5分钟步长一条数据会被12个窗口覆盖。这意味着如果每秒有1万条商品行为数据那么窗口聚合的实际计算量相当于每秒12万条。这个放大倍数在数据量大的时候是致命的。那有没有办法避免重复计算有但要看场景。一种思路是用增量聚合代替全量聚合。Flink的窗口聚合分两种全量聚合apply/process把窗口内所有数据缓存起来窗口触发时一次性计算和增量聚合reduce/aggregate每来一条数据就更新一次累加器窗口触发时直接输出累加器结果。增量聚合不会把原始数据全存下来内存占用小得多而且计算是流式的不会在窗口触发时产生计算尖峰。另一种思路是换一种实现方式。如果只是要最近一小时TopN其实不一定非要用滑动窗口。可以用一个更轻量的方案维护一个按时间分桶的滚动窗口比如每分钟一个桶查询时把最近60个桶合并起来算TopN。这种方式把12倍放大降到了1倍存储查询时合并在很多实时数仓场景里反而更实用。Flink SQL里的HOP窗口本质上还是滑动窗口但如果用TUMBLE加外部查询合并就能实现类似效果。我个人的经验是数据量不大每秒几千条以内时直接用滑动窗口最省事数据量大或者对延迟敏感时优先考虑增量聚合再不行就上分桶合并的方案。这个案例作为入门我们先用标准的滑动窗口把流程跑通后面再聊优化。2.3 时间语义选事件时间还是处理时间时间语义是流处理里另一个绕不开的选择。处理时间Processing Time用的是机器本地时钟简单、低延迟但结果不确定——同样的数据换个时间跑结果可能不一样。事件时间Event Time用的是数据自带的时间戳结果可复现但需要处理乱序数据得配水位线Watermark。对于最近一小时热门商品这种业务指标我强烈建议用事件时间。原因很实际商品行为数据从产生到进入Flink中间可能经过埋点上报、消息队列、网络传输延迟几秒到几分钟都正常。如果用处理时间那么最近一小时统计的其实是最近一小时到达Flink的数据而不是最近一小时真实发生的行为两者在数据延迟时会差很多。用事件时间配合水位线容忍一定程度的乱序结果才贴近业务真实情况。水位线的设置也有讲究。设得太紧乱序数据会被丢弃统计偏少设得太松窗口触发延迟大看板更新不及时。常见做法是设一个比最大预期乱序时间略大的值比如允许迟到30秒水位线就设forBoundedOutOfOrderness(Duration.ofSeconds(30))。这个值需要根据实际数据链路的延迟分布来调没有万能答案。3. DataStream API路线手写窗口聚合与TopN的完整链路这一节我们用DataStream API把整个流程实现一遍。选DataStream API而不是直接上SQL是因为它能让你看清每一步的数据形态变化对理解Flink的运行机制帮助很大。等你把这条链路走通了再看SQL版本会觉得豁然开朗。3.1 数据源与事件时间字段的提取假设我们的商品行为数据来自消息队列格式是JSON字段包括用户ID、商品ID、行为类型点击、加购、下单、时间戳。第一步是定义POJO并接入数据源。public class UserBehavior { public String userId; public String itemId; public String behavior; public long timestamp; public UserBehavior() {} public UserBehavior(String userId, String itemId, String behavior, long timestamp) { this.userId userId; this.itemId itemId; this.behavior behavior; this.timestamp timestamp; } }接入数据源后关键一步是分配时间戳和生成水位线。在Flink 1.12之后推荐用WatermarkStrategy统一处理DataStreamUserBehavior stream env .addSource(new FlinkKafkaConsumer(user_behavior, new JSONDeserializationSchema(), properties)) .assignTimestampsAndWatermarks( WatermarkStrategy.UserBehaviorforBoundedOutOfOrderness(Duration.ofSeconds(30)) .withTimestampAssigner((event, recordTimestamp) - event.timestamp) );这里有个坑我踩过forBoundedOutOfOrderness的参数是最大允许乱序时间不是水位线延迟时间。它的含义是我允许数据迟到最多30秒Flink会把水位线设为当前最大事件时间减去30秒。如果你设成0那就是不允许任何乱序来一条迟到的就丢一条统计会明显偏少。3.2 用增量聚合统计每个商品的窗口内热度接下来是窗口聚合。我们要统计的是每个商品在窗口内的行为次数所以先按商品ID分组再开滑动窗口然后用增量聚合累加计数。DataStreamTuple2String, Long itemCount stream .keyBy(event - event.itemId) .window(SlidingEventTimeWindows.of(Time.minutes(60), Time.minutes(5))) .aggregate(new AggregateFunctionUserBehavior, Long, Long() { Override public Long createAccumulator() { return 0L; } Override public Long add(UserBehavior 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; } }) .map(count - Tuple2.of(count.getKey(), count.getValue())) .returns(Types.TUPLE(Types.STRING, Types.LONG));这段代码有几个细节值得说。第一keyBy之后开窗口窗口是作用在每个key上的也就是说每个商品独立维护自己的窗口状态。第二用aggregate而不是apply是因为我们只需要计数增量聚合足够而且内存友好。第三merge方法在滑动窗口里其实用不到滑动窗口不会合并但AggregateFunction接口要求实现写个简单的相加即可。这里要提醒一个常见误区窗口聚合的输出是每个窗口每个key一条结果。也就是说如果一小时内有1000个商品被访问过那么每个5分钟窗口会输出1000条记录每个商品一条。这些记录是分散的不是排好序的TopN。要得到TopN还需要后续处理。3.3 把分散的计数聚成TopN榜单拿到每个商品的窗口计数后下一步是按窗口分组取每个窗口内计数最高的N个商品。这里要用到windowAll或者按窗口结束时间分组再排序。一种常见做法是先把窗口结果转成窗口结束时间商品计数的格式然后按窗口结束时间keyBy用KeyedProcessFunction维护一个定时器在窗口结束时排序输出TopN。但更简单的做法是用windowAll配合ProcessAllWindowFunctionitemCount .windowAll(SlidingEventTimeWindows.of(Time.minutes(60), Time.minutes(5))) .process(new ProcessAllWindowFunctionTuple2String, Long, String, TimeWindow() { Override public void process(Context context, IterableTuple2String, Long elements, CollectorString out) { ListTuple2String, Long list new ArrayList(); for (Tuple2String, Long e : elements) { list.add(e); } list.sort((a, b) - Long.compare(b.f1, a.f1)); StringBuilder sb new StringBuilder(); sb.append(窗口结束时间: ).append(context.window().getEnd()).append(\n); for (int i 0; i Math.min(10, list.size()); i) { sb.append(Top).append(i 1).append(: ) .append(list.get(i).f0).append( 热度: ).append(list.get(i).f1).append(\n); } out.collect(sb.toString()); } }) .print();这段代码能跑但有个性能隐患windowAll会把所有key的数据汇聚到一个并行度为1的算子数据量大时会成为瓶颈。生产环境更推荐用keyBy加KeyedProcessFunction的方式让每个窗口的TopN计算分散到多个并行实例上。不过作为入门案例windowAll足够直观先跑通再优化。注意windowAll的并行度强制为1这是它的设计限制不是配置问题。数据量大时务必换方案。4. Flink SQL路线同样的需求用声明式写法怎么落地DataStream API把每一步都摊开给你看好处是透明代价是代码量大。Flink SQL走的是另一条路你只描述要什么不描述怎么做引擎帮你优化执行计划。同样的需求SQL版本可能只要十几行。4.1 建表与时间属性定义SQL版本的第一步是建表。这里的关键是把事件时间字段声明为时间属性这样后续才能用窗口函数。CREATE TABLE user_behavior ( user_id STRING, item_id STRING, behavior STRING, ts BIGINT, event_time AS TO_TIMESTAMP_LTZ(ts, 3), WATERMARK FOR event_time AS event_time - INTERVAL 30 SECOND ) WITH ( connector kafka, topic user_behavior, properties.bootstrap.servers localhost:9092, properties.group.id topn_group, scan.startup.mode latest-offset, format json );这里event_time是一个计算列由ts转换而来同时被声明为水位线字段。WATERMARK FOR event_time AS event_time - INTERVAL 30 SECOND等价于DataStream里的forBoundedOutOfOrderness(Duration.ofSeconds(30))。注意时间戳单位TO_TIMESTAMP_LTZ(ts, 3)里的3表示毫秒精度如果你的时间戳是秒级要改成0。4.2 滑动窗口聚合与TopN的SQL表达有了表定义滑动窗口聚合就是一句GROUP BY加HOPCREATE VIEW item_count_view AS SELECT item_id, HOP_START(event_time, INTERVAL 5 MINUTE, INTERVAL 60 MINUTE) AS window_start, HOP_END(event_time, INTERVAL 5 MINUTE, INTERVAL 60 MINUTE) AS window_end, COUNT(*) AS cnt FROM user_behavior GROUP BY item_id, HOP(event_time, INTERVAL 5 MINUTE, INTERVAL 60 MINUTE);HOP(event_time, 步长, 窗口大小)就是滑动窗口的SQL表达参数顺序是时间字段、滑动步长、窗口大小别写反了。HOP_START和HOP_END分别取窗口的起止时间方便后续按窗口分组。接下来是TopN。Flink SQL用ROW_NUMBER()配合OVER窗口来实现CREATE VIEW topn_view AS SELECT * FROM ( SELECT item_id, window_start, window_end, cnt, ROW_NUMBER() OVER (PARTITION BY window_end ORDER BY cnt DESC) AS rn FROM item_count_view ) WHERE rn 10;PARTITION BY window_end表示按窗口结束时间分组ORDER BY cnt DESC按热度降序rn 10取前10名。这个写法比DataStream的windowAll优雅得多而且Flink SQL会自动做优化不会强制单并行度。4.3 SQL版本相比DataStream省了什么、又牺牲了什么SQL版本最大的优势是开发效率。同样的需求DataStream要写上百行JavaSQL可能二十行搞定而且不用关心算子并行度、状态后端这些底层细节。对于实时数仓场景SQL几乎是首选。但它也有代价。第一灵活性受限。比如你想在TopN里加一个排除某个类目的逻辑SQL要改视图DataStream改一行filter就行。第二调试困难。SQL的执行计划是引擎生成的出了问题只能看explain不像DataStream能打断点。第三状态管理不透明。SQL的窗口状态由引擎管理你很难精细控制TTL和清理策略。我的建议是能用SQL就用SQL遇到SQL表达不了的复杂逻辑再退回DataStream。两者不是对立的实际项目里经常混用——SQL做ETL和聚合DataStream做复杂的自定义处理。5. 状态、并行度与迟到数据那些跑通之后才会遇到的问题代码能跑通只是第一步。真正上线之后你会发现状态膨胀、并行度不匹配、迟到数据丢失这些问题一个接一个冒出来。这一节聊聊这些跑通之后的坑。5.1 滑动窗口的状态为什么会越积越多滑动窗口的状态管理是个容易被低估的问题。因为一条数据属于12个窗口Flink需要为每个窗口维护独立的聚合状态。如果窗口大小是60分钟、步长5分钟那么任意时刻都有12个窗口处于打开状态每个窗口每个key都有一份累加器。假设有10万个商品key每个累加器占100字节那么12个窗口乘以10万key乘以100字节大约是120MB。看起来不大但如果key的数量涨到千万级或者累加器更复杂比如存了TopN列表状态就会迅速膨胀。控制状态的手段有几个。第一设置状态TTL。窗口状态在窗口触发后其实就没用了但Flink默认不会立即清理可以配置StateTtlConfig让状态在窗口结束后一段时间自动过期。第二用增量聚合而非全量聚合。全量聚合会把窗口内所有原始数据存下来状态量是增量聚合的几十倍。第三考虑分桶合并方案。前面提到的每分钟一个桶、查询时合并的思路状态量只有滑动窗口的十二分之一。5.2 并行度和keyBy的配合有什么讲究并行度设置直接影响吞吐和延迟。keyBy之后每个key的数据会被路由到固定的并行实例上同一个key的窗口状态只存在于一个实例。这意味着并行度不能超过key的基数否则会有实例空转。滑动窗口场景下我一般会这样估算并行度先看峰值QPS再看单个并行实例能处理多少。假设单实例能处理5000条每秒峰值QPS是5万那么并行度至少设10。但别忘了滑动窗口有12倍放大所以实际并行度要按60万条每秒来算也就是120。这个数字可能超出集群资源这时候就得回到增量聚合或者分桶方案来降负载。还有一个细节windowAll的并行度是1如果你在TopN阶段用了windowAll那么不管前面并行度设多高最后都会汇聚到一个算子。这是很多新手觉得明明设了并行度还是慢的原因。5.3 迟到数据到底该丢还是该等迟到数据是事件时间语义下必然面对的问题。数据迟到几分钟甚至几小时都有可能怎么处理取决于业务对准确性的要求。Flink提供了几种处理迟到数据的机制。水位线决定了窗口何时触发水位线之前的数据如果迟到默认会被丢弃。allowedLateness允许窗口在触发后再等一段时间期间到达的迟到数据会重新触发窗口计算。sideOutput可以把超时迟到的数据输出到侧流单独处理。.window(SlidingEventTimeWindows.of(Time.minutes(60), Time.minutes(5))) .allowedLateness(Time.minutes(2)) .sideOutputLateData(lateOutputTag)这段配置的含义是窗口触发后再等2分钟期间到达的迟到数据会重新计算并输出更新结果超过2分钟的数据进入侧流。侧流的数据可以写入另一个topic后续人工核对或补偿。我的经验是看板类指标可以容忍少量迟到数据丢失设个合理的allowedLateness就行对账类指标必须保证不丢那就得用侧流兜底。没有绝对的对错看业务能接受什么。6. 两条路线的选型建议与几个实测心得聊完DataStream和SQL两条路线最后说说我实际项目里的选型逻辑和一些踩坑心得。6.1 什么时候用DataStream什么时候用SQL这个问题没有标准答案但有几条经验可以参考。逻辑简单、以聚合和join为主的场景优先SQL开发快、维护成本低。需要自定义状态、复杂事件处理、多流精细控制的场景用DataStream。团队SQL能力强于Java的尽量往SQL靠反之亦然。还有一个现实因素SQL的版本兼容性比DataStream好。DataStream API在不同大版本之间经常有废弃和重构SQL相对稳定。如果你的项目要长期维护SQL的升级成本更低。6.2 窗口大小和步长的参数怎么定60分钟配5分钟不是唯一答案具体参数要看业务。窗口大小决定了热度的时间范围步长决定了看板刷新频率。步长越小看板越实时但计算量越大。我见过有团队把步长设成1分钟结果集群扛不住最后改成5分钟。一个实用的调参思路是先确定业务能接受的最大延迟再反推步长。如果运营说5分钟内看到更新就行那步长设5分钟如果说要准实时那可能得设1分钟同时接受更高的资源消耗。窗口大小则看业务对热度的定义一小时是常见选择也有用30分钟或2小时的。6.3 几个我踩过的具体坑第一个坑是时间戳单位搞错。Kafka里的时间戳有的是秒级有的是毫秒级TO_TIMESTAMP_LTZ的精度参数写错水位线就会差几个数量级窗口永远不触发。排查这类问题时先打印几条数据的时间戳看看量级。第二个坑是水位线设得太紧导致数据大量丢失。我一开始把乱序容忍设成0结果统计结果比实际少了一大截。后来改成30秒数据就对上了。水位线的值要根据实际链路的延迟分布来定可以先设大一点观察迟到数据量再收紧。第三个坑是忘了设置状态TTL导致状态爆炸。滑动窗口的状态如果不清理长时间运行后checkpoint会越来越大最后拖垮整个作业。加上TTL配置后状态量稳定在一个可控范围。第四个坑是TopN的排序在数据倾斜时不准。如果某个商品特别热门它的数据会集中到一个并行实例导致该实例负载过高甚至影响排序结果的及时性。解决办法是对热门key做加盐打散或者用两阶段聚合。6.4 这个案例还能怎么扩展把基础版本跑通之后可以往几个方向扩展。加维度按商品类目、店铺、地区分别统计TopN这需要多级keyBy或者SQL的多维group by。加指标不只统计点击还统计加购率、转化率这需要多流join或者自定义聚合函数。加去重同一用户对同一商品的重复点击只算一次这要用到去重状态。加告警某个商品热度突然飙升时触发告警这要配合CEP或者自定义的阈值检测。每一个扩展方向都是一次对Flink能力的深入练习。我的建议是不要贪多先把5分钟统计1小时TopN这一个场景吃透把状态、并行度、迟到数据这几个核心问题都摸清楚再往下一个场景走。流处理的很多坑是相通的一个场景踩透了换场景也能举一反三。最后分享一个我调试这类作业的小习惯先在本地用一个小数据集把逻辑跑通确认窗口触发时间和输出结果符合预期再上集群跑真实数据。本地调试时把并行度设成1输出每个窗口的起止时间和聚合结果一眼就能看出窗口边界对不对。等逻辑没问题了再逐步调大并行度、接真实数据源。这个习惯帮我省下了大量在集群上反复试错的时间。