简介基于Flink商品实时推荐系统的完整项目资料包面向计算机相关专业学生、老师及企业开发者可支撑毕业设计、课程设计、作业演示和初期立项参考也是大数据实时计算入门的实用素材。项目代码已在测试环境运行通过获得导师指导认可答辩评审分达到95分。资源共47个文件以34份源码文件为主体配合SQL脚本、HBase建表语句、Kafka模拟数据、属性配置和XML配置及说明文档压缩包仅247KB却完整覆盖推荐系统从数据接入、实时计算、特征处理到结果存储的核心链路目录结构明晰方便按模块阅读和二次开发。目前已有58人学习下载适合具备一定大数据基础的同学深入掌握Flink实时推荐细节也可直接用于课程设计或毕业设计并在此基础上扩展新的推荐功能。1. 基于 Flink 的商品实时推荐系统从用户点击到推荐结果的这条大数据流水线如果把用户刚看完商品、首页就出现相似推荐这件事拆开看它背后是一条完整的实时计算流水线。这套基于 Flink 的商品实时推荐系统正是把这条流水线做成了可运行的工程Kafka 接收行为日志Flink 做实时 ETL、热度统计和协同过滤召回结果落到 HBase再由后端接口输出到页面。它不是那种只有几个类的教学 demo而是带完整工程结构、独立 flink-2-hbase 模块、模拟数据和 SQL 脚本的毕设级项目特别适合正在准备大数据方向毕业设计或者想把 Flink 流计算真正落地跑一遍的同学。2. 架构设计与技术选型一条真实数据链路的分层方式拿到这份资源后我先做了一件事不看代码看目录。工程结构往往比代码更能说明一个项目的成熟度。这套推荐系统没有把逻辑堆在一个模块里而是拆出了 ETL、热度计算、协同过滤、HBase 写入、接口服务几个职责边界每一块都能单独运行、单独调试。2.1 整体架构从日志埋点到推荐接口的五层链路整个系统可以分成五层每一层都对应工程里的一个具体模块。数据从用户行为产生到最后变成接口返回的推荐列表中间经过的每一跳都是推荐系统里真实存在的环节。flink-recommend-system-main ├── pom.xml # 父工程聚合各模块 ├── data # 模拟数据集csv │ ├── user.csv # 用户基础数据 │ ├── product.csv # 商品分类与属性数据 │ └── user_behavior.csv # 用户行为日志点击/加购/下单 ├── sql │ ├── init.sql # MySQL 建表脚本 │ └── prepare.sql # 抽样数据初始化脚本 ├── flink-2-hbase # Flink 写 HBase 独立模块 │ ├── pom.xml # 单独打成可提交的 jar │ └── src/main/java/com/recommend/hbase │ ├── HBaseSink.java # 自定义 Sink批量写入推荐结果 │ └── HBaseConfig.java # HBase 连接与表结构配置 └── src └── main ├── java/com/recommend │ ├── etl/UserBehaviorETL.java # 行为日志解析与清洗 │ ├── hot/HotProductJob.java # 实时热门商品统计 │ ├── recall/CollaborativeFilter.java # 基于物品的协同过滤召回 │ └── service/RecommendService.java # 推荐结果查询接口 └── resources ├── application.yml # 后端服务配置 └── log4j2.xml先说结论这个目录结构最大的优点是职责边界清晰。父工程用 pom 聚合data 和 sql 单独存放说明作者一开始就把数据准备和代码逻辑分开了。实际跑的时候模拟数据从 data 目录读入 KafkaMySQL 里的商品和用户表由 sql 脚本初始化Flink 消费 Kafka 日志后做清洗和计算再把结果写入 HBase最后 RecommendService 对外提供查询接口——整条链路闭环了。我自己拆项目时有个习惯先画一条数据流再在代码里找流上的每个节点。这个项目的对应关系是user_behavior.csv → Kafka topic → Flink 实时计算 → HBase 表 recommend_result → SpringBoot 接口 → Web 展示。如果你准备拿它做毕设答辩 PPT 里直接放这条链路图比贴代码直观得多。2.2 技术选型为什么是 Flink Kafka HBase Redis 这套组合这套技术栈是典型的实时推荐系统标配。不是因为它流行而是每层都有不可替代的理由。组件承担职责选型理由Kafka行为日志接入与削峰高吞吐、可回放Flink 消费它天然带 offset 管理Flink实时 ETL、窗口统计、协同过滤事件时间处理、精准一次语义、状态管理能力强HBase存储用户推荐结果和用户画像列式存储适合按 userId 随机读写天然支持稀疏字段Redis热门榜单和实时特征缓存毫秒级读取配合 HBase 做二级缓存MySQL存储商品、用户、订单基础数据初始化数据和后端服务查询需要为什么用 Flink 而不是 Spark Streaming这是这个项目里值得展开的一个点。Spark Streaming 本质上是微批一批数据的延迟通常在秒级甚至更高而 Flink 是真正的流式计算每条数据到了就处理搭配事件时间Event Time和 watermark能处理乱序到达的行为日志。这套系统里有滑动窗口统计热门商品Flink 的滑动窗口 API 是原生支持的窗口触发、迟到数据处理都封装得很完整——用微批框架做同样的事窗口边界的维护要自己写不少代码。Flume Spark 这种组合在课程设计里很常见但真实场景下行为日志量一大Flume 的 channel 就会成瓶颈。Kafka Flink 的问题是 Kafka 的吞吐上限和 Flink 的背压机制能不能配合好——Flink 消费速度跟不上时会自动把压力反馈给 Kafka不会直接把内存打爆。这个项目的定位是实时推荐系统选这套组合本身就是架构上最合理的答案。另外注意 pom.xml 里把 flink-2-hbase 单独拆成一个模块这是工程上的一个实用技巧。Flink 任务提交到集群或本地环境时HBase 客户端的依赖经常和 Flink 自带的依赖冲突常见的如 guava、protobuf把这部分独立成模块、单独打包能避免提交大 fat jar 时把冲突也带上去。3. 核心模块拆解一个行为从进 Kafka 到落 HBase 的完整链路这一章是关键。推荐系统的核心不在后端的 CRUD而在 Flink 里那几条实时计算链路。我从源码里拆出了四个重点行为日志 ETL、热度窗口统计、HBase Sink、协同过滤召回。把这四个点吃透这套系统你就掌握了一半。3.1 行为日志 ETL把点击、加购、下单规整成一张明细表Kafka 里存的原始行为日志是字符串格式里带着逗号分隔的字段。Flink 任务启动后第一步就是消费这个 topic把字符串解析成 Java 对象同时把脏数据过滤掉。DataStreamString sourceStream env.addSource(new FlinkKafkaConsumer( user-behavior, // topic 名称 new SimpleStringSchema(), // 反序列化方式 kafkaProps // bootstrap.servers / group.id )); DataStreamUserBehavior behaviorStream sourceStream .map(new MapFunctionString, UserBehavior() { Override public UserBehavior map(String line) throws Exception { // 日志格式userId,productId,behaviorType,eventTime String[] fields line.split(,); if (fields.length 4) { // 字段缺失的直接丢弃避免脏数据污染后续计算 return null; } return new UserBehavior( Long.parseLong(fields[0].trim()), // 用户 ID Long.parseLong(fields[1].trim()), // 商品 ID fields[2].trim(), // 行为类型 Long.parseLong(fields[3].trim()) // 行为时间戳毫秒 ); } }) // 过滤掉解析失败产生的 null确保下游拿到的都是有效对象 .filter(Objects::nonNull);这段代码逻辑不复杂但有三个细节值得注意。第一解析时做了字段长度校验不到 4 个字段的直接在 map 里返回 null再用 filter 过滤掉——这是流式 ETL 最常见也最有效的脏数据处理方式比在算子内部抛异常再捕获要干净得多。第二parseLong 之前做了 trim因为生产环境日志里经常混入空格不 trim 就会 NumberFormatException 导致整个任务失败。第三行为类型保留为字符串方便后续扩展——新增一种行为不需要改结构体只要下游逻辑识别新类型就行。实际开发中我会在这个环节多做一步把无法解析的脏数据单独写入一个 Kafka 死信 topic而不是直接丢掉。这样任务运行期间发现问题还能回溯原始日志。这份源码没做这个属于可以二次开发加的点。3.2 实时热度统计滑动窗口给商品打综合热度分热度推荐是最简单也最常用的召回策略把一段时间内被点击、加购、下单最多的商品推给用户。这个模块用 Flink 的滑动窗口实现每 1 分钟计算一次最近 10 分钟的热度并且用 watermark 处理乱序日志。DataStreamHotProduct hotStream behaviorStream // 指定事件时间提取和 watermark 策略 .assignTimestampsAndWatermarks( WatermarkStrategy.UserBehaviorforBoundedOutOfOrderness(Duration.ofSeconds(10)) .withTimestampAssigner((event, timestamp) - event.getEventTime()) ) // 按商品 ID 分组每个商品独立计算 .keyBy(UserBehavior::getProductId) // 10 分钟窗口1 分钟滑动一次 .window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(1))) // 先做增量聚合避免窗口内缓存大量原始数据 .aggregate(new HotAggregate(), new HotWindowResult());重点说两个参数的设置逻辑。Time.minutes(10) 是窗口长度Time.minutes(1) 是滑动步长——意思是拿最近 10 分钟的数据算热度每过 1 分钟刷新一次榜单。步长越小榜单实时性越好但计算量也越大推荐系统里通常取分钟级。forBoundedOutOfOrderness(Duration.ofSeconds(10)) 表示允许事件迟到最多 10 秒超过这个时间的数据会被丢弃10 秒这个值是根据日志上报链路延迟定的行为日志从 App 上报到 Kafka 一般延迟在几秒内设置 10 秒比较安全。HotAggregate 是增量聚合函数核心逻辑是给行为加权点击计 1 分、加购计 2 分、下单计 3 分。这里有个值得一提的设计——热度分不是简单 count而是按行为价值加权否则用户随便点一下就跟花真金白银下单一个权重榜单会被无意义的点击刷屏。权重值来源是纯粹的运营经验我在实际项目里还会把收藏行为加进去甚至做时间衰减越靠近当前时间的行为权重越高避免一个上个月爆火、现在过气的商品长期霸榜。3.3 flink-2-hbase 模块自定义 Sink 批量化写入推荐结果前面计算的推荐结果要存起来供接口查询。这个模块是独立工程 flink-2-hbase核心是继承 RichSinkFunction 实现一个写 HBase 的 Sink。public class HBaseSink extends RichSinkFunctionRecommendResult { private transient Connection connection; private transient BufferedMutator mutator; Override public void open(Configuration parameters) throws Exception { // open 只在任务启动时调用一次这里适合创建连接池 Configuration conf HBaseConfiguration.create(); conf.set(hbase.zookeeper.quorum, localhost:2181); connection ConnectionFactory.createConnection(conf); // BufferedMutator 支持缓冲区批量写入避免每条数据都走一次 RPC BufferedMutatorParams params new BufferedMutatorParams( TableName.valueOf(recommend_result)) .writeBufferSize(1024 * 1024 * 8); // 8MB 缓冲区 mutator connection.getBufferedMutator(params); } Override public void invoke(RecommendResult result, Context context) throws Exception { // RowKey 直接用 userId查询时 get 单行即返回 Put put new Put(Bytes.toBytes(result.getUserId())); // 推荐列表以 JSON 字符串存入单列 put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(recommend_list), Bytes.toBytes(JSON.toJSONString(result.getRecommendList()))); mutator.mutate(put); } Override public void close() throws Exception { // 最后强制刷空缓冲区避免任务结束时数据还滞留在内存 if (mutator ! null) { mutator.flush(); } } }这段代码里有三个地方能看出作者不是第一次写 HBase 应用。第一是用了 BufferedMutator 而不是 table.put()Flink 的写入频率高每条数据单独 put 会产生大量 RPC用 BufferedMutator 攒够 8MB 再批量写吞吐能提升好几个量级。第二是把连接建立放在 open() 里——RichSinkFunction 的 open 在任务生命周期里只调用一次连接复用是连接池的基本操作如果放在 invoke 里每条数据都创建连接任务跑几分钟就会把 Zookeeper 连接数打满。第三是 close() 里做了 flush防止任务正常结束时缓冲区里还有未落盘的数据这个不写会有数据丢失。RowKey 的设计也值得讲。以 userId 作为 RowKey查询推荐结果时就是一次精确 get——这是 HBase 最理想的行键。但实际生产里用户量大了顺序递增的 userId 会导致写热点集中在少数 Region所以一般会做成散列形式比如反转字符串或者加盐。这份源码直接用的原始 userId课设和毕设场景下完全够用。3.4 协同过滤召回用会话窗口建立商品共现矩阵热度榜解决的是大家都在买什么协同过滤解决的是和你品味相似的人还买什么。这个模块实现的是基于物品的协同过滤核心是实时更新商品共现矩阵——两个商品被同一个用户在同一时间窗口内浏览过它们之间就记一次共现。// 只取浏览行为参与共现计算加购和下单的数据量太小共现矩阵会非常稀疏 DataStreamTuple2Tuple2Long, Long, Integer coOccurStream behaviorStream .filter(behavior - view.equals(behavior.getBehaviorType())) // 按用户分组每个用户的浏览行为形成自己的序列 .keyBy(UserBehavior::getUserId) // 5 分钟的会话窗口窗口内出现的商品两两互相记一次共现 .window(TumblingEventTimeWindows.of(Time.minutes(5))) .process(new CoOccurrenceProcessFunction());这个实现的巧妙之处在于用 5 分钟的 Tumbling 窗口近似一个浏览会话。同一个用户 5 分钟内看了哪些商品就认为这些商品之间是相关的——比如用户看了一部手机又看了它的保护壳这两个商品就产生共现。这个设计避免了把用户所有历史行为都算进来既降低计算量又让相关更有实时意义一个用户一个月前看的商品和现在看的商品未必真相关。CoOccurrenceProcessFunction 的输出是一个三元组((商品A, 商品B), 共现次数)。下游再按 商品B 聚合所有共现分数排序取 TopN就是商品B 的相似商品列表。后续做推荐时用户近期看过商品A就从 商品A 的相似列表里取前几名推给用户。这个召回策略有个先天的坑就是冷启动问题——新商品没有任何共现记录永远不会被推荐到。第四行会讲这个项目的兜底方案如何解决算是我拆代码时发现的比较欣赏的一个设计。4. 运行部署从环境准备到推荐接口返回结果很多同学下载项目后卡在跑不起来原因不是代码有问题而是环境版本不匹配。这一章我按自己实际跑通这套系统的顺序把从空环境到接口响应的完整过程写出来。4.1 环境准备版本搭配和最小配置这套系统依赖挺多但单机开发环境不需要搭完整集群Kafka、HBase、Flink 全部单机模式即可。我本地跑通用的版本搭配是JDK 8务必要用 8这个项目不是针对高版本编译的、Maven 3.6、Kafka 2.4.1、HBase 1.4.9、Flink 1.9.2。这几个版本互相兼容是我验证过的新手最不容易翻车的组合。软件建议版本部署模式用途JDK1.8本机安装编译和运行基础Maven3.6本机安装依赖管理和打包Kafka2.4.1单机模式行为日志消息队列Zookeeper3.4.xKafka 自带Kafka 内置Kafka 和 HBase 都依赖HBase1.4.9单机 standalone推荐结果存储Flink1.9.2本地模式执行实时计算任务启动顺序有讲究先启动 Zookeeper再启动 Kafka然后启动 HBase。因为 HBase 也依赖 Zookeeper 做协调。如果你电脑上之前装过别的 Zookeeper 版本注意统一版本否则 HBase 起不来会出现很诡异的连接报错。# 1. 启动 ZookeeperKafka 内置脚本 bin/zookeeper-server-start.sh config/zookeeper.properties # 2. 启动 Kafka bin/kafka-server-start.sh config/server.properties # 3. 启动 HBasestandalone 模式 bin/start-hbase.sh # 4. 验证 HBase 是否正常 bin/hbase shell listHBase 的版本坑是这里最值得提前说的不要用 2.x 以上版本。Flink 1.9 内置的 HBase connector 是针对 1.4 版本编译的换成 2.x 会报 AbstractMethodError属于典型的代码没问题、环境不兼容的问题血泪经验。4.2 数据准备模型建表与模拟数据导入运行前要把基础数据准备好。sql 目录里 init.sql 是 MySQL 的建表脚本prepare.sql 是数据初始化脚本——商品表、用户表、订单表都会建好prepare.sql 还会往表里插入一批模拟数据。data 目录下的 csv 是行为日志的模拟源头。# 初始化 MySQL假设本地 MySQL 已启动 mysql -uroot -p sql/init.sql mysql -uroot -p sql/prepare.sql # 创建 Kafka topic bin/kafka-topics.sh --create --topic user-behavior \ --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1 # 读取 csv 模拟数据写入 Kafka模拟行为日志实时产生 cat data/user_behavior.csv | bin/kafka-console-producer.sh \ --broker-list localhost:9092 --topic user-behaviorKafka 这里有一个非常重要的配置细节partitions 设成 1。如果你的 Flink 任务并行度设成了 4但 Kafka topic 只有 1 个分区那实际只有 1 个并发在消费数据其他并行度空闲——这不算错误但会让新手误以为任务出 bug 了。这节课设项目阶段建议分区数并行度后面到生产环境再按吞吐量调整。csv 数据的格式要和 ETL 解析代码对齐——逗号分隔的 userId,productId,behaviorType,eventTime。在真实项目里这一环节应该由前端埋点 SDK 完成把用户行为实时上报到 Kafka这里用脚本模拟了数据源。跑通之后想拿真实数据分析只需要换个 producer链路不用动。4.3 启动步骤按依赖顺序拉起整条链路环境就绪、数据通路测试完之后按照依赖顺序启动编写好的 Flink 任务。核心是先让 Flink 任务跑起来消费 Kafka再启动后端服务读取 HBase 里的推荐结果。# 1. 编译整个工程 mvn clean package -DskipTests # 2. 提交热度计算任务本地模式运行直接 java -jar 即可 java -jar flink-2-hbase/target/flink-2-hbase-1.0.jar # 3. 执行主模块中的推荐计算任务 mvn compile exec:java -Dexec.mainClasscom.recommend.etl.UserBehaviorETL # 4. 启动 SpringBoot 推荐接口服务 mvn spring-boot:run启动顺序的建议理由是HBase 必须在 Flink 任务之前启动因为 Sink 在 open() 阶段就要建立连接连接失败任务会直接报错退出。而 Kafka 必须在 Flink 之前是没得选的前提因为 Kafka consumer 启动时就要连接 broker。按 4.1 → 4.2 → 4.3 的顺序走不会出错。有一个经验如果用的是本地模式运行 Flink 任务但 pom 里声明了 scope 为 provided 的依赖全部会漏掉——provided 意味着运行时不提供由容器提供。本地执行 main 方法时没有 Flink 容器帮加载这些依赖就会报 ClassNotFoundException。如果遇到这个报错把 provided 临时改成 compile 或者用 exec-maven-plugin这是本地跑 Flink 常见的隐蔽问题。4.4 验证链路日志、HBase 和接口三层确认启动完成后不能只看进程还活着就说跑通了要分三层验证。# 1. 看 Flink 任务日志确认消费到数据 tail -f flink-recommend.log # 看到 INFO: Current watermark: 16xxxxxxxxx 说明 watermark 在推进 # 看到 Sink: recommend_result: 20 说明有结果写入 # 2. 查 HBase 表确认推荐结果真的写进去了 bin/hbase shell scan recommend_result, {LIMIT 5} # 能看到 cf:recommend_list 列有内容 # 3. 调推荐接口确认后端能读出来 curl http://localhost:8080/recommend?userId1001 # 返回 JSON 数组[商品ID: 2034, 1022, 1873, ...]第一次跑通常会在第二步卡住——HBase scan 结果为空。绝大多数情况是 Kafka producer 脚本读 csv 时数据还没发送完Flink 窗口又还没触发。注意热度窗口是 10 分钟长度、1 分钟滑动所以至少要等 1 分钟才可能看到第一次窗口输出。去喝口水再回来看如果还是没有再检查 watermark 是否推进。这一整套从日志到存储再到接口的三层验证也是我排查大数据链路问题时的固定习惯。5. 避坑记录这套 Flink 推荐系统最容易翻车的四个位置跑通这套系统不难但跑的过程中有几个位置非常容易翻车。我拆这套源码时把踩过的坑整理成大块机制问题每条都按现象 → 原因 → 解决写清楚。5.1 HBase 数据写不进去报 RegionTooBusyException 或连接被拒绝现象启动 Flink 任务后日志里看到 HBase 写入抛异常任务反复重启。原因两个高频元因组合出现。一是 hbase.zookeeper.quorum 配置的是 localhost:2181但 HBase 实际用的 Zookeeper 端口被 Kafka 自带的 Zookeeper 占用了。Kafka 内置 Zookeeper 默认端口 2181HBase 默认连的也是 2181——两边抢同一个端口HBase 的 meta 信息根本读不到。二是写入频率太高单条 put 导致 Region 频繁 flush。解决给 HBase 换独立端口比如在 hbase-site.xml 里设置 hbase.zookeeper.property.clientPort2182然后把 HBaseConfig.java 里的 quorum 也改成 2182写入端改用 BufferedMutator源码里 HBaseSink 已经处理了这一点但如果你是自己在二次开发中新写的 Sink很容易走回单条 put 的老路。5.2 Flink 任务重启后Kafka 重复消费导致推荐结果翻倍现象任务重启一次HBase 里同一个 userId 的推荐列表出现重复商品热度分值异常偏高。原因Flink 默认没开 Checkpoint 时是 At-Least-Once 语义。任务重启后Kafka offset 从上次提交的位置继续读已经处理过的数据会再处理一遍。这个项目里体现得最明显的就是窗口聚合——同一批日志被算进两个窗口热度分数就翻倍了。解决在 Flink 任务入口开启 Checkpoint并将 Kafka 消费者语义改成 Exactly-Once。代码里加上 env.enableCheckpointing(5000) 并设置 checkpoint 的存储路径同时给 FlinkKafkaConsumer 指定 setStartFromLatest() 或者维护好 offset。从那以后我每次启动任务都强制走一遍确认 Checkpoint 已开启这个检查这个习惯帮我避免了很多日志对不上的问题。5.3 事件时间窗口一直不触发榜单数据迟迟不出现象任务启动后日志显示 watermark 一直停留在初始值热度窗口 10 分钟了也没输出。原因窗口使用 EventTime但没有正确设置时间戳提取器和 watermark 生成策略。源码里用 assignTimestampsAndWatermarks 指定了 from 字段但如果你改了行为日志格式、或者模拟数据里的时间戳字段不是毫秒而传了秒级值watermark 推进就会异常慢甚至不动。解决确认 eventTime 是毫秒级时间戳如 1688888888000。若是秒级10 位数字要在 withTimestampAssigner 里乘以 1000。排查时可以用一个临时日志算子打印 watermark 的变化我一般会加一个这样的调试算子map 里输出每条数据的时间戳和当前 watermark跑一分钟就能定位是时间戳格式问题还是 watermark 策略配错的问题。5.4 冷启动阶段推荐结果为空协同过滤一个商品都召回不出来现象系统刚上线HBase 里只写了热度榜数据协同过滤的相似商品表 5 分钟会话窗口一直聚合不出有效结果。原因协同过滤依赖行为数据积累如果模拟数据量小或者用户行为时间跨度长5 分钟窗口内用户浏览的商品太少共现矩阵稀疏到接近零。解决做两层召回。第一层是热度榜兜底——无论有没有个性化结果都返回热门商品列表第二层才是个性化协同过滤。这个项目本身就是这么设计的HBase 表里热度榜和个性化推荐各占一行接口层先读写推荐发现为空就优雅降级读热门榜。验证时可以将模拟数据的时间戳集中在一个时间段内并写一个循环脚本连续发送数据就能看到协同过滤结果逐步被填充。冷启动是推荐系统的经典难题并不是代码 bug提前把兜底策略写进代码里是上线前就该做的事。6. 二次开发把推荐策略换成你自己的召回算法这章写给想基于这套源码做毕设、课设或练手的同学。项目现有链路是完整的但推荐策略相对基础正好留出了改造空间。我给出三个可以动手的方向以及每个方向改完后怎么验证。6.1 替换召回策略从人工加权热度换成用户兴趣衰减模型当前代码里 HotAggregate 用的是固定权重点击 1 分、加购 2 分、下单 3 分。这个模型的问题在于不考虑时间——用户上个月的加购和今天的点击拥有同等权重。改造时可以引入时间衰减指数分数 行为权重 × e^(-Δt / halfLife)其中 Δt 是行为时间距离窗口末端的时间差halfLife 是衰减半衰期。代码改动点只在 HotAggregate 的累加逻辑里把原来的 count 变成加权求和不动窗口和 Sink 结构。// 改造后的热度累加逻辑在原有 weight 基础上乘时间衰减 double score behaviorWeight * Math.exp(-(windowEnd - eventTime) / (60.0 * 1000 * 10));这个衰减半衰期设 10 分钟意味着 10 分钟前的一次点击到了窗口末端的权重只剩原来的 36.8%。热点榜单会更敏锐地反映正在流行而不是历史总热度。改完以后调接口观察连续对同一商品发送加购行为1 分钟内该商品在榜单的排名上升速度应该明显比旧逻辑快。6.2 增加相似商品池的召回宽度现网效果差的真正原因当前协同过滤窗口设的 5 分钟实现的是会话内共现。有个常见的问题——用户买东西不一定在 5 分钟内完成决策用户昨天看的一部手机今天又看了它的配件这两件商品因为隔天而错过了共现机会。改造方法比较直观把 TumblingEventTimeWindows.of(Time.minutes(5)) 改成 SlidingEventTimeWindows.of(Time.hours(24), Time.hours(1))也就是以 1 小时为步长滑动、24 小时跨度的窗口来做会话共现。滑动窗口存的状态会变大但配合 RocksDB 状态后端开销可控。// 把 5 分钟会话窗口改成 24 小时滑动窗口捕捉跨天再看一眼的行为 .window(SlidingEventTimeWindows.of(Time.hours(24), Time.hours(1)))改完验证时拉长时间范围构造数据给同一用户在第 1 小时和第 10 小时分别浏览两个商品旧代码不会产生共现记录新代码会产生。你可以把这个作为二次开发后的功能自测用例跑通了说明窗口语义理解到位了。6.3 改造效果怎么验证一份可操作的验证清单无论改召回还是改排序验证都要分两层走。第一层是离线日志验证在 data 目录里构造一组已知期望的模拟数据比如用户 A 浏览了商品 X期望推荐结果里出现与 X 共现过的商品 Y跑完任务后直接查 HBase 表对比结果。第二层是接口侧回归启动全链路后用 curl 请求推荐接口关注两个指标——推荐列表是否非空、列表的多样性是否比改造前更高。如果你准备拿它做毕设答辩时展示这个旧逻辑 vs 新逻辑的对比输出比讲一百页原理都有效。这套系统我断断续续改过三版最深的一个体会是Flink 项目调试的玄学往往不在代码本身而在状态管理。窗口算子的状态、Kafka offset 的状态、HBase 的缓冲区这三样东西只要有一个没对齐表现就会特别诡异。从那以后我每次改完代码都强制走一遍清空 Kafka 从最新消费 清空 HBase 表 重新灌模拟数据三部曲确保我看到的输出是本次代码的直接结果而不是旧状态的残留。希望这份拆解能帮你在自己的节点上少走几步弯路尽快跑通这套基于 Flink 的商品实时推荐系统。本文还有配套的精品资源点击获取