Flink实时推荐系统全链路解析：从Kafka到HBase与Redis

简介：这套基于 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 单独算一个推荐分数那么简单，它是一条完整的数据管道：前端埋点采集用户行为，行为消息进 Kafka，Flink 负责实时清洗、聚合、关联特征，再把结果落到 HBase 和 Redis，最后由推荐服务读取结果返回给页面。整体链路可以拆成四段。

第一段是数据接入。用户浏览商品、点击详情、加购、收藏这些动作，通过埋点 SDK 上报，落到 Kafka 的 topic 里。这个 topic 通常按行为类型分，也可以混在一个 topic 里靠 type 字段区分。第二段是 Flink 核心计算。Flink 从 Kafka 消费原始日志，解析成结构化对象，做去重、补全维度信息，再按照用户维度或商品维度开窗口计算。实时热门榜、用户最近浏览序列、商品共现关系，都是在这一步产出的。第三段是结果存储。计算出来的实时推荐候选集写进 Redis，因为推荐服务要毫秒级读取；行为明细和历史聚合结果写进 HBase，用于离线回刷和模型训练。第四段是服务输出。推荐接口从 Redis 拿候选集，再做规则过滤、多样性打散，最后拼装成前端需要的 JSON。

资料包里的 data 目录就是为第一段准备的。常见做法是先启动 Kafka，建好 topic，然后用命令行脚本把 data 下的样例数据灌进去。建 topic 的命令一般长这样。

# 创建 user_behavior topic，3 个分区，副本因子 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 里。

组件和职责可以用一张表看明白。

这套设计把“写多读少”和“读多写少”两种负载拆开，各自用合适的存储，这是整个推荐系统能实时跑起来的基础。

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 这个子模块。整体结构并不复杂，用一张表可以看清单个目录的职责。

pom.xml 是这个模块的地基。Flink 工程最大的坑是依赖版本冲突，特别是连接器版本和 Flink 主版本不一致，导致运行时各种 NoClassDefFound 和 AbstractMethodError。常见的做法是引入一个 BOM 统一管理版本，子模块里不写具体版本号。打开 flink-2-hbase 的 pom.xml，核心依赖大概是这样。

<!-- Flink 主依赖，BOM 统一管理版本 --> <dependencyManagement> <dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-java</artifactId> <version>${flink.version}</version> </dependency> </dependencies> </dependencyManagement> <!-- Kafka 连接器，消费行为日志 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka_2.12</artifactId> <version>${flink.version}</version> </dependency> <!-- HBase 连接器，写入行为明细和共现矩阵 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-hbase-2.2_2.12</artifactId> <version>${flink.version}</version> </dependency> <!-- RocksDB 状态后端，替代默认内存状态 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-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.12，Flink 的 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"); DataStream<String> source = env.addSource(new FlinkKafkaConsumer<String>( "user_behavior", // topic 名 new SimpleStringSchema(), // 消息体是纯 JSON 字符串 props )); // 第一步：解析 JSON 成 UserBehavior 对象 DataStream<UserBehavior> behaviors = source .map(new JsonToBehaviorMapper()) .assignTimestampsAndWatermarks(new BehaviorWatermark()); // 第二步：按用户分组，开 10 分钟滑动窗口，步长 1 分钟 DataStream<List<UserBehavior>> 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 RichSinkFunction<List<UserBehavior>> { 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(List<UserBehavior> 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 RichSinkFunction<List<UserBehavior>> { @Override public void invoke(List<UserBehavior> behaviors, Context context) throws Exception { // 计算用户候选推荐集合，这里简化成最近浏览商品的相似商品 List<String> recommendItems = buildCandidateList(behaviors); // 写入 Redis，60 秒过期，过期后由 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 集群。环境清单大概如下。

版本匹配是这一节最大的坑。Flink 1.13 配 HBase 2.2 和 Kafka 2.x 是常见的稳定组合，如果 Flink 升到 1.16，Kafka 连接器的包名和部分 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

如果本地没有 HDFS，checkpoints 目录可以换成 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 算子背压级别为 HIGH，Kafka 的消费组 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_servers='localhost:9092', value_serializer=lambda v: json.dumps(v).encode('utf-8') ) # 逐行回放 data 目录下的行为数据 with open('data/user_behavior.json', encoding='utf-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 有没有变化。这套流程看着简单，真能帮你把“本地能跑”和“链路对了”这两件事分清楚。希望帮到你。

本文还有配套的精品资源，点击获取