简介这套基于 Flink 的商品实时推荐系统资料包面向大数据与推荐系统方向的在校学生、教师及企业开发者定位于毕业设计、课程设计或项目立项演示完整覆盖实时推荐链路的工程实现。压缩包内共有 47 个文件以 34 个 Scala 源码文件为主体包含 Flink 消费 Kafka 模拟数据、HBase 存储与查询等核心逻辑另有 SQL 建表/数据脚本、HBase 建表语句、properties 与 xml 配置、Kafka 模拟数据脚本以及说明文档结构清晰便于按模块阅读和复用整体约 247KB轻量易部署。目前已有 58 人学习下载。项目为个人高分项目已通过导师审核与答辩评审代码经测试运行成功读者可获得完整源码、详细说明文档、数据模拟与配置脚本可在此基础上修改扩展或直接用于毕设、课设适合从入门到进阶学习 Flink 实时推荐链路的搭建与扩展。1. 为什么实时推荐绕不开 Flink一套能直接跑的流式全链路某电商大促当晚行为日志像洪水一样涌进来基于离线批处理的推荐结果要等半小时才刷新用户看到的推荐永远慢半拍点击率掉得让人心慌。后来把链路换成 Flink 大数据实时计算从用户点击到推荐结果更新压到秒级点击率肉眼可见地涨了。这份基于 Flink 的商品实时推荐系统资料就是一套能直接跑的流式推荐全流程。它不只给代码连详细文档、测试数据、SQL 初始化脚本、HBase 落库逻辑都整理在一个压缩包里。做毕设、课设或上手实时推荐都值得照着拆一遍。适合正在学大数据流式计算、想搞懂推荐系统落地链路的人。2. 先理清架构和数据流组件选型与实时链路全景2.1 一条完整的实时推荐链路长什么样在拆这套资源之前先把链路摆出来。实时推荐不是让 Flink 单独算一个推荐分数那么简单它是一条完整的数据管道前端埋点采集用户行为行为消息进 KafkaFlink 负责实时清洗、聚合、关联特征再把结果落到 HBase 和 Redis最后由推荐服务读取结果返回给页面。整体链路可以拆成四段。第一段是数据接入。用户浏览商品、点击详情、加购、收藏这些动作通过埋点 SDK 上报落到 Kafka 的 topic 里。这个 topic 通常按行为类型分也可以混在一个 topic 里靠 type 字段区分。第二段是 Flink 核心计算。Flink 从 Kafka 消费原始日志解析成结构化对象做去重、补全维度信息再按照用户维度或商品维度开窗口计算。实时热门榜、用户最近浏览序列、商品共现关系都是在这一步产出的。第三段是结果存储。计算出来的实时推荐候选集写进 Redis因为推荐服务要毫秒级读取行为明细和历史聚合结果写进 HBase用于离线回刷和模型训练。第四段是服务输出。推荐接口从 Redis 拿候选集再做规则过滤、多样性打散最后拼装成前端需要的 JSON。资料包里的 data 目录就是为第一段准备的。常见做法是先启动 Kafka建好 topic然后用命令行脚本把 data 下的样例数据灌进去。建 topic 的命令一般长这样。# 创建 user_behavior topic3 个分区副本因子 1单机环境 kafka-topics.sh --bootstrap-server localhost:9092 \ --create --topic user_behavior \ --partitions 3 --replication-factor 1 # 查看 topic 是否创建成功 kafka-topics.sh --bootstrap-server localhost:9092 --list分区数建议直接设成 3后面 Flink 作业并行度对齐这个数字能少踩一个反压的坑。单机学习环境副本因子只能设 1如果设 3 而 Kafka 只有一个 broker会一直报元数据异常。把样例数据灌进去时用 kafka-console-producer 一行一行发或者直接写一个生产者脚本控制发送速率。2.2 为什么行为明细存 HBase、推荐结果存 Redis很多第一次做实时推荐的人会问行为数据为什么不能直接放 MySQL原因有两个一个是写入吞吐一个是查询模式。用户行为是持续产生的高吞吐追加流MySQL 在千万级写入下扛不住而且行为数据要按用户维度回看、按时间范围扫描这种查询模式在 MySQL 里会频繁扫大表。HBase 的宽表模型天然适合这个场景rowkey 设计成用户 ID 加时间戳一个用户的所有行为连续存储扫描一批数据只需要一次 range 查询。Redis 承担的角色完全不同。推荐服务是在用户请求路径上的延迟要求通常小于 50 毫秒HBase 虽然能扛写入但读路径要走网络和磁盘稳定读延迟在几十毫秒量级直接让推荐服务读 HBase 不太合适。Redis 把 Flink 算好的 TopN 推荐结果以 userId 为 key 存成 JSON 字符串推荐服务一次 GET 就能拿全。这里有一个关键取舍Redis 只存最终推荐结果不存原始日志原始日志始终在 HBase 里。组件和职责可以用一张表看明白。组件存储内容写入方读取方核心特点Kafka原始行为日志埋点 SDKFlink高吞吐消息队列削峰填谷HBase行为明细、商品特征、共现矩阵FlinkFlink 离线任务、推荐服务海量数据随机读写横向扩展Redis用户实时推荐结果Flink推荐服务毫秒级读取TTL 自动过期Flink无状态存储负责计算数据源接入HBase、Redis事件驱动、窗口计算、状态管理这套设计把“写多读少”和“读多写少”两种负载拆开各自用合适的存储这是整个推荐系统能实时跑起来的基础。2.3 推荐算法在流式任务里的两种常见变体拿到这套资源里面带的算法并不复杂常见的是实时热门榜加协同过滤的流式变体。先看热门榜怎么实时算。Flink 里用滑动窗口对商品点击事件计数窗口长度 10 分钟滑动步长 1 分钟每分钟输出一次当前热门商品 TopN。这个逻辑用一个 window 加一个 TopN 函数就能实现不涉及复杂的模型训练。协同过滤在流式场景下和离线版本不太一样。离线版本是定期扫全量行为日志算商品共现矩阵生成每个用户的推荐列表再落到线上。流式版本把共现矩阵改成增量更新用户每产生一次点击行为就把该商品和用户最近浏览过的其它商品之间的共现次数加一矩阵本身存放在 HBase 的一张表里。用户请求推荐时从矩阵里取该用户最近 N 个商品的相似商品合并去重后输出。把离线协同过滤搬到流式至少要完成四步操作。第一步从行为流里提取用户的最近行为序列通常用一个滑动窗口加状态来维护。第二步在窗口触发时对序列里的商品两两组合更新共现矩阵对应 HBase 里的 increment 操作。第三步计算用户对候选商品的偏好分数常见做法是用户最近行为的商品权重乘共现次数的累加。第四步结果写 Redis同时设置 TTL。这套逻辑在资源里已经完整实现读代码时重点看窗口的触发条件和 HBase 的 increment 写法。3. 把工程代码拆开看flink-recommend-system-main 的模块与核心任务3.1 Maven 工程结构与 pom 关键依赖解开压缩包后核心工程目录是 flink-recommend-system-main里面又分了 flink-2-hbase 这个子模块。整体结构并不复杂用一张表可以看清单个目录的职责。目录或文件作用flink-2-hbase/pom.xml子模块的 Maven 依赖管理flink-2-hbase/src/main/javaFlink 作业源码flink-2-hbase/src/main/resources作业运行配置文件data样例行为数据模拟用户点击流sqlHBase 表结构、Redis 初始化相关 SQL 或脚本README.md项目说明和启动步骤pom.xml 是这个模块的地基。Flink 工程最大的坑是依赖版本冲突特别是连接器版本和 Flink 主版本不一致导致运行时各种 NoClassDefFound 和 AbstractMethodError。常见的做法是引入一个 BOM 统一管理版本子模块里不写具体版本号。打开 flink-2-hbase 的 pom.xml核心依赖大概是这样。!-- Flink 主依赖BOM 统一管理版本 -- dependencyManagement dependencies dependency groupIdorg.apache.flink/groupId artifactIdflink-java/artifactId version${flink.version}/version /dependency /dependencies /dependencyManagement !-- Kafka 连接器消费行为日志 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-kafka_2.12/artifactId version${flink.version}/version /dependency !-- HBase 连接器写入行为明细和共现矩阵 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-hbase-2.2_2.12/artifactId version${flink.version}/version /dependency !-- RocksDB 状态后端替代默认内存状态 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-statebackend-rocksdb_2.12/artifactId version${flink.version}/version /dependency依赖里有几个细节值得注意。flink-connector-hbase-2.2 后面的 2.2 是 HBase 的版本号它对应的是 HBase 2.x 系列的连接器如果你的环境是 HBase 1.x运行时会报 HBase 配置类找不到。另一处是 Scala 版本号 _2.12Flink 的 Kafka 客户端依赖 Scala如果工程用的 Scala 版本是 2.11这里要换成 _2.11。RocksDB 状态后端加进来之后默认内存状态后端就不会再让大状态把堆内存撑爆这个坑后面会专门讲。3.2 从 Kafka 接入行为数据的 Flink 任务骨架核心作业入口是一个标准的 Flink DataStream 程序。它的职责是从 Kafka 消费行为日志把 JSON 字符串转成 Java 对象按用户分组后进入窗口计算。代码骨架如下。public class RecommendJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); // 开启 checkpoint间隔 5 秒精确一次语义 env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE); // 设置状态后端为 RocksDB避免堆内存 OOM env.setStateBackend(new EmbeddedRocksDBStateBackend()); Properties props new Properties(); props.setProperty(bootstrap.servers, localhost:9092); props.setProperty(group.id, recommend-flink-group); DataStreamString source env.addSource(new FlinkKafkaConsumerString( user_behavior, // topic 名 new SimpleStringSchema(), // 消息体是纯 JSON 字符串 props )); // 第一步解析 JSON 成 UserBehavior 对象 DataStreamUserBehavior behaviors source .map(new JsonToBehaviorMapper()) .assignTimestampsAndWatermarks(new BehaviorWatermark()); // 第二步按用户分组开 10 分钟滑动窗口步长 1 分钟 DataStreamListUserBehavior userActions behaviors .keyBy(UserBehavior::getUserId) .window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(1))) .process(new UserActionWindowFunction()); // 第三步写入 HBase 和 Redis userActions.addSink(new HBaseBehaviorSink()); userActions.map(new RecommendListBuilder()).addSink(new RedisRecommendSink()); env.execute(flink-recommend-system); } }关键逻辑在中间两段。assignTimestampsAndWatermarks 是为行为数据分配事件时间戳和水位线推荐系统里用户行为天然有乱序必须用事件时间而不是处理时间否则晚到的日志会被窗口丢弃。SlidingEventTimeWindows 是 10 分钟窗口、1 分钟滑动一次这样每秒输出的推荐结果会随时间平滑变化不会出现整点跳变的效果。UserActionWindowFunction 是一个 ProcessWindowFunction它在窗口内收集该用户的所有行为排序后形成最近行为序列。HBaseBehaviorSink 把这个序列展开逐个更新商品共现矩阵RedisRecommendSink 则把算好的 TopN 列表写入 Redis。看这段代码时要抓住两个边界。第一个是水位线watermark 设得太小晚到的数据会丢推荐结果不完整设得太大窗口触发延迟高实时性变差。常见的做法是允许 5 到 10 秒的乱序延迟同时加 allowedLateness 再兜底。第二个是窗口状态每个用户的最近行为序列存在 Flink 的 keyed state 里如果用户量很大RocksDB 状态后端是必须的。3.3 特征聚合与结果写出窗口计算、HBase 和 Redis sink窗口函数处理完结果要落到两个存储里。HBase 写的是一张共现矩阵表结构是 rowkey 为商品 A ID列族 cf 下有一列列名是商品 B ID列值是共现次数。这个设计的好处是可以直接用 HBase 的 increment 操作原子累加不需要先读后写。核心代码段如下。public class HBaseBehaviorSink extends RichSinkFunctionListUserBehavior { private Connection conn; Override public void open(Configuration parameters) throws Exception { // 作业启动时初始化 HBase 连接 Configuration hbaseConfig HBaseConfiguration.create(); hbaseConfig.set(hbase.zookeeper.quorum, localhost:2181); conn ConnectionFactory.createConnection(hbaseConfig); } Override public void invoke(ListUserBehavior behaviors, Context context) throws Exception { Table table conn.getTable(TableName.valueOf(item_similarity)); // 对窗口内的商品序列做两两组合更新共现次数 for (UserBehavior current : behaviors) { for (UserBehavior other : behaviors) { if (!current.getItemId().equals(other.getItemId())) { Increment increment new Increment( Bytes.toBytes(current.getItemId())); increment.addColumn( Bytes.toBytes(cf), Bytes.toBytes(other.getItemId()), 1L); table.increment(increment); } } } table.close(); } }这段代码的逻辑是双重循环生成商品对。比如用户在一个窗口里浏览了 A、B、C 三个商品那就会产生 A-B、A-C、B-A、B-C、C-A、C-B 六个共现对每个共现对加一。HBase 的 increment 是原子操作并发场景下不会丢失计数。写入 Redis 的方式也很直接以 userId 为 key把推荐列表序列化成 JSON 写入同时设置 5 分钟过期时间过期后 Flink 下一次窗口触发又会重新写入保证推荐内容是新鲜的。public class RedisRecommendSink extends RichSinkFunctionListUserBehavior { Override public void invoke(ListUserBehavior behaviors, Context context) throws Exception { // 计算用户候选推荐集合这里简化成最近浏览商品的相似商品 ListString recommendItems buildCandidateList(behaviors); // 写入 Redis60 秒过期过期后由 Flink 重新计算写入 try (Jedis jedis new Jedis(localhost, 6379)) { String key recommend:user: behaviors.get(0).getUserId(); String value objectMapper.writeValueAsString(recommendItems); jedis.setex(key, 60, value); } } }Redis Sink 的优化点是连接池。每条数据都 new Jedis 会建立大量连接连接成了瓶颈。项目代码里如果没做池化改造时用 JedisPool 就能明显提升吞吐。另外 setex 的过期时间不要设太长推荐结果本身是实时的设太短会导致缓存频繁穿透设 60 秒到 5 分钟之间比较合适。3.4 data 与 sql 目录的初始化建表与造数据资源里的 sql 目录一般包含 HBase 表结构的初始化脚本。实时推荐至少需要两张表一张存用户行为明细一张存商品共现矩阵。HBase 的建表命令是 create和关系型数据库不同它建表时必须指定列族不能先建表再改列族。# 行为明细表列族 cf保留 3 个版本 create user_behavior, {NAME cf, VERSIONS 3, COMPRESSION SNAPPY} # 共现矩阵表预分区 20 个 Region避免单 Region 热点 create item_similarity, {NAME cf, COMPRESSION SNAPPY}, {NUMREGIONS 20, SPLITALGO HexStringSplit}第一张表存行为明细rowkey 设计成 userId 加时间戳这样同一个用户的行为按时间顺序连续存储查询时按 rowkey range 扫一段即可。第二张表存共现矩阵因为商品 ID 是字符串直接做主键容易出现热点写入预分区能把这个风险降低。data 目录下的样例数据通常是 JSON 格式一行一个事件字段包括 userId、itemId、behaviorType、timestamp。把数据灌进 Kafka 之后Flink 作业一旦启动就能立刻消费并产出结果不用等真实流量。4. 把配置与部署跑通从本地模式到集群提交的几个关键点4.1 本地快速验证的最小环境清单这套资源要在本地跑起来先确定一件事是起全套集群还是只要作业能提交。如果只是为了看代码和跑通作业用 Flink 的本地模式加单机 Kafka、单机 HBase、单机 Redis 就够了不需要 Hadoop 集群。环境清单大概如下。组件版本建议用途JDK1.8 或 11编译和运行Maven3.6 以上工程编译打包Kafka2.x 或 3.x消息队列HBase2.x行为明细存储Redis5.x 以上推荐结果缓存Flink1.13 到 1.16流式计算框架版本匹配是这一节最大的坑。Flink 1.13 配 HBase 2.2 和 Kafka 2.x 是常见的稳定组合如果 Flink 升到 1.16Kafka 连接器的包名和部分 API 有变化代码里 import 路径可能要跟着改。启动顺序也有讲究先起 HBase再起 Kafka最后提交 Flink 作业。HBase 启动后要等它完全就绪用 status 命令确认 RegionServer 在线了再继续否则 Flink 作业的 HBase Sink 会初始化失败。4.2 HBase 预分区建表rowkey 设计与热点规避很多人在本地玩的时候建 HBase 表不打预分区数据量小看不出来问题一旦灌入几百万行真实行为数据热点 Region 就会让写入延迟急剧上升。预分区的思路是建表时把 rowkey 空间切成 20 个区间每个区间一个 Region写入时数据分散到不同 Region而不是全部压在一个节点上。建表语句在 3.4 已经给了关键在 SPLITALGO 的选择上。HexStringSplit 是把 rowkey 按十六进制字符串区间均匀切分适合 rowkey 是字符串前缀的场景。如果 rowkey 是数字自增 ID比如纯时间戳适合用 UniformSplit 或手动指定切分点。这个资源里的 rowkey 设计包含了用户 ID 和时间戳用户 ID 往往是随机字符串用 HexStringSplit 基本能打散热点。还有个常用技巧是 rowkey 加盐把用户 ID 取哈希后取前一位作为前缀拼上原 ID写 HBase 前先判断盐值查询时枚举所有前缀再合并结果。这样写性能和读性能都能保证代价是代码里要多一层枚举逻辑。4.3 状态后端与 checkpoint 配置Flink 作业默认状态后端是内存窗口逻辑维持每人每天的行为序列状态量一旦涨起来JobManager 和 TaskManager 的堆内存都会告急。把状态后端切到 RocksDB状态会落到本地磁盘只有热数据留在内存这是实时推荐这类大状态作业的标配。配置方式有两种一种写死在代码里一种写在 flink-conf.yaml 里。# flink-conf.yaml 关键配置项 state.backend: rocksdb state.checkpoints.dir: hdfs:///flink/checkpoints state.checkpoints.num-retained: 10 state.backend.incremental: true如果本地没有 HDFScheckpoints 目录可以换成 file:///tmp/flink-checkpoints但要注意单机文件路径在 Flink 重启后可能残留大量旧 checkpoint定期清理或者直接用 state.checkpoints.num-retained 控制保留数量。incremental 增量 checkpoint 对 RocksDB 效果明显每次都全量备份会拖慢整个作业开启增量之后备份体积小很多。提交作业时如果代码里已经设置了 StateBackend配置文件里的设置会被代码覆盖两者不一致时以代码为准。4.4 并行度与 Kafka 分区让窗口和数据一起对齐Flink 作业的并行度设置是玄学也是硬功夫。并行度设高了资源浪费设低了Kafka 消费不过来背压一路传导到 Source。这里有一个强约束Kafka Source 的并行度最好和 topic 分区数一致。Kafka 的一个分区只能被同一消费组里的一个消费者线程消费如果 Source 并行度 6 而分区数是 3会有 3 个线程闲住数据还是只有 3 个线程在消费。提交命令里用 -p 指定并行度常见做法是先用默认并行度跑一个样例数据集观察 Kafka 的 lag 有没有持续上涨再根据结果调整。# 提交作业并行度 3和 Kafka topic 分区数保持一致 flink run -m yarn-per-job -d \ -p 3 \ -c com.example.RecommendJob \ flink-recommend-system-1.0.jar-m yarn-per-job表示以 per-job 模式提交到 YARN如果只是本地验证去掉 -m 参数直接 flink run 即可。-d表示后台分离运行不带这个参数的话终端关闭作业就跟着停了。每个算子也可以单独调并行度比如 Redis Sink 的并行度可以设为 2不让它和 Source 一样高因为 Redis 写入受连接数限制太高反而造成连接竞争。5. 避坑与排查让实时任务稳定跑的六条血泪记录5.1 现象Flink 作业刚启动就报 Kafka 连接失败任务无限重启任务日志里频繁出现 Connection refused指向的地址是 broker 的 hostname 而不是 IP。本地单机环境最容易遇到这个问题。原因是 Kafka 的 server.properties 里配置了 advertised.listeners这个地址在 Flink 所在节点解析不了比如 адрес 配置成了 kafka-node1但 /etc/hosts 里没有这个条记录。解决办法是把 advertised.listeners 改成 localhost:9092或在 Flink 所在机器的 /etc/hosts 里加上 broker 主机名和 IP 的映射。改完 Kafka 配置后要重启 Kafka 才会生效别只改不重启那样会让问题更隐蔽。5.2 现象HBase 写入越来越慢RegionServer 日志显示写放大严重写入量不到百万级别但 HBase 的 flush 频繁触发Region 数量不均衡部分 Region 写入 QPS 是其它 Region 的几十倍。原因是 rowkey 设计带上了纯递增时间戳前缀所有写入都落在最后一个 Region 上前面的 Region 完全没有写入压力热 Region 达到容量上限后又触发 split进一步拖慢整体写入。解决办法是给 rowkey 加盐把用户 ID 的哈希前缀放在时间戳前面写入压力摊到所有 Region。如果现有数据已经写入还要用 HBase 的 snapshot 导出、重新建表、再导入的流程来修复没有后悔药可吃。5.3 现象Kafka 消费 lag 持续上涨反压指标一路飘红Flink Web UI 里可以看到 Source 算子背压级别为 HIGHKafka 的消费组 lag 不降反升。常见原因是 Redis Sink 写入太慢逐条写 Redis 没有批量处理Flink 的缓冲区被 sink 拖住背压向上传导。也可能是 Sink 并行度和 Source 不一致导致的线程瓶颈。解决办法分两步先看 Web UI 里哪个算子反压最严重如果确认是 Redis sink改成窗口攒批写入或者用 JedisPool 替代每次 new Jedis。再把 Sink 并行度调低到 2减少连接竞争。改完观察 lag 是否回落。5.4 现象TaskManager 堆内存持续上涨最终 OOM 崩溃本地环境用默认状态后端跑大窗口作业行为数据一多内存就爆。原因是没有切换 RocksDB 状态后端keyed state 全部留在堆内存里。用户数量越大每个用户的行为序列越多堆内存增长越快。解决办法是 pom 里加 flink-statebackend-rocksdb 依赖代码里设置状态后端为 RocksDB配置增量 checkpoint。换完之后要注意磁盘空间RocksDB 会占用本地磁盘单节点环境要预留至少和状态量等量的磁盘空间。5.5 现象推荐结果一直不更新Redis 里的 key 还是旧数据Flink 作业正常运行Kafka 也有数据进入但 Redis 里推荐 key 的 TTL 到了之后没有新值写入。原因大概率是窗口水位线没推进会窗口不触发所有 sink 都不执行。行为数据的时间戳如果比当前时间晚或者乱序严重而又没设置合理的 watermark窗口会一直等数据。解决办法是检查数据里的 timestamp 字段是不是时间戳格式单位是秒还是毫秒要和代码里一致。另一个常见原因是 Kafka topic 没有新数据进入用 kafka-console-consumer 验证一下 topic 里到底有没有消息比对着 Flink 日志猜要快得多。5.6 现象本地模式运行秒过提交到集群直接 NoSuchMethodError同一个 jar 本地跑没问题丢到集群后各种类冲突最常见的是 flink-shaded 相关类版本不匹配。原因是本地环境的 classpath 是 IDE 拉好的依赖集群环境用的是 Flink 自带的依赖两者一旦版本不同就会起冲突。解决办法是打包时用 maven-shade-plugin把工程依赖的第三方类 shade 进去并对容易被冲突的包做 relocation。另外一个排查思路是检查 pom 里是否引入了和 Flink 自带重复的依赖比如 flink-core、flink-streaming-java 这类标记为 provided 即可。6. 把端到端延迟验证做起来模拟点击流回放检查推荐更新能不能真正证明这套推荐系统是实时的取决于你如何验证端到端延迟。只看 Flink 作业运行正常不够因为 most 坑都藏在链路中间。我常用的技巧是用脚本按固定节奏回放行为数据然后盯着 Redis 里的推荐 key 看它多久刷新一次。回放脚本不复杂一行一读一发送。import json import time from kafka import KafkaProducer # 连接 Kafka发送行为日志 producer KafkaProducer( bootstrap_serverslocalhost:9092, value_serializerlambda v: json.dumps(v).encode(utf-8) ) # 逐行回放 data 目录下的行为数据 with open(data/user_behavior.json, encodingutf-8) as f: for line in f: record json.loads(line.strip()) producer.send(user_behavior, record) # 每 0.3 秒发一条模拟真实在线用户点击 time.sleep(0.3)回放之后用 redis-cli 直接看某个用户的推荐 key 是否在更新。命令长这样。# 先拿到样例数据里的某个 userId替换下面的占位符 redis-cli GET recommend:user:10001 # 隔 60 秒再查一次对比两次结果是否变化 sleep 60 redis-cli GET recommend:user:10001两次结果如果一样说明窗口还没触发或者 watermark 卡住如果内容有变化说明整条链路是通的。接下来再量一下延迟在发数据的时间点和 Redis key 更新时间点之间插入时间戳用脚本判断窗口输出延迟是否在预期范围。滑动窗口是 1 分钟滑动一次理论上最多延迟 1 分钟能看到更新。验证完链路之后还能顺手测试一个边界把回放脚本的发送间隔调大比如从 0.3 秒改成 10 秒观察推荐 key 是否还按窗口准时刷新。这个测试能验证作业不是靠“碰巧有新数据才触发”的伪实时而是真正由事件时间水位线驱动的。之后再改任何逻辑比如调整窗口大小、改 Redis 过期时间都先跑一遍回放脚本对比推荐的更新节奏。后来我每次做实时推荐项目都会强制走一遍端到端回放验证先起 Kafka再起 Flink 作业回放行为数据最后盯 Redis 里的推荐 key 有没有变化。这套流程看着简单真能帮你把“本地能跑”和“链路对了”这两件事分清楚。希望帮到你。本文还有配套的精品资源点击获取
