简介基于Flink的商品实时推荐系统项目包适合大数据开发与推荐系统学习者用于解决实时商品热度统计、用户画像构建及个性化推荐排序等核心问题。资源共109个文件、压缩包3.74MB以68个Java源文件为主另有SQL建表脚本、HBase建表语句、XML配置、Properties与YML配置文件、Kafka模拟数据脚本及HTML展示页面覆盖从数据接入、缓存存储到推荐计算和前端展示的完整链路。系统采用Flink统计商品热度并写入Redis缓存同时分析用户日志生成画像标签、实时记录存入HBase用户发起推荐请求时系统基于用户画像对热度榜重排序并融合协同过滤与标签推荐两个模块为每个商品补充关联产品最终返回个性化推荐列表。目前已有130人学习。通过源码可掌握Flink流处理、Redis与HBase集成、协同过滤及标签推荐算法的工程落地方法适合进阶实战参考。1. 实时推荐不是算法竞赛而是特征时效性的竞赛用户在商品详情页刚滑走的第 5 秒离线推荐系统还在等当天 T1 日志清洗完成而基于 Flink 的商品实时推荐系统已经把“刚刚浏览过这件商品”写成了下一轮召回的特征。这套方案要解决的核心问题是把用户行为从小时级延迟压缩到秒级让每一次点击都能真实影响下一次推荐内容。它不引入什么新模型而是把从埋点到特征、再到召回排序的整条链路做成实时闭环。适合已经有埋点、数据仓库和离线推荐但特征延迟仍停留在小时级以上的团队新手可以把这里当作 flink 实时计算的第一个完整项目熟手则能看到状态管理和连接器选型的真实边界。2. 整体架构与选型Flink 在推荐链路里的准确位置实时推荐不是把离线推荐的代码换个执行引擎就跑而是要把原来“按天算好写回表”的特征改成“持续流入、持续更新”的流式特征。Flink 在这里承担的是特征计算层而不是完整的推荐系统。先把这层位置摆正后面每一步选型才有依据。2.1 推荐系统的四层链路埋点、特征、召回、排序推荐系统看起来是一个在线服务实际是一条多层接力链路。第一层是埋点采集App 或 Web 页面把用户行为曝光、点击、加购、下单实时上报到 Kafka第二层是流式特征计算这是 Flink 的主战场按用户维度聚合行为序列按物品维度计算关联关系第三层是召回在线服务根据用户当前行为从特征库里取候选集第四层是排序把候选集按预估点击率排序后输出到页面。中间第二层之所以必须用流式计算而不是离线批处理是因为召回和排序都要读“上一秒”的特征。离线推荐把特征打成 T1 宽表在线服务读到的永远是一天前的偏好实时推荐把这个宽表拆成不断追加的流式特征用户只要在窗口期内发生了行为特征就被更新下游立刻可见。商品信息这类变化不频繁的维度数据常见做法是用 Flink CDC Pipeline 把业务库的变更直接同步到特征存储避免定时全量同步带来的延迟黑洞。需要提醒的是flink cdc 安装部署本身不难但 pipeline 模式对连接器版本兼容性比较敏感数据量大之前先花半天在测试环境跑通整条链路比上了生产再排查划算得多。2.2 为什么选 Flink 而不是 Spark Streaming 或自研管道落到选型团队里最常见的争论是 Spark Streaming 也能消费 Kafka为什么非要 Flink。在实时推荐这个场景里差距主要体现在三个点事件时间语义、状态管理和容错恢复。埋点日志在 Kafka 里只要稍微积压基于处理时间的窗口统计就会把 10 点的事件算进 10 点 05 分的窗口推荐结果看起来在更新实际上是错位的。Flink 的事件时间加水印机制可以按日志里的时间戳重新对齐行为序列不会因为中间链路抖动而错乱。对比维度Spark StreamingFlink时间语义以处理时间为主事件时间支持有限事件时间 水印原生支持状态管理窗口内状态窗口结束即释放ValueState / ListState TTL容错恢复批式重跑存在重复计算分布式快照可做到精确一次端到端延迟秒级到分钟级秒级以内这张对比表基本就是拍板依据。实时推荐属于典型的“状态密集 窗口密集”场景每个用户的行为序列是持续累积的状态每个窗口的共现统计依赖精确的时间对齐这两点正好都是 Flink 的长项。部署层面flink 安装配置到部署按标准集群流程走就可以但要记住环境搭建只决定作业能不能跑起来窗口和状态的设计才决定推荐效果好坏。2.3 数据流与角色划分从 Kafka 到 Redis 的接力在推荐场景里Flink 作业通常拆成两个角色。一个是用户特征作业消费行为日志按 userId 分组用状态维护最近 N 分钟的行为序列产出的结果写入 Redis另一个是物品关联作业消费同一份行为日志计算物品之间的共现关系产出物品相似度集合。两个作业可以合并成一个 DAG也可以分开部署我一般建议分开。二者的吞吐特征和故障影响面不一样拆开后可以独立调优并行度、独立重启排查问题时也不需要把整条链路停掉。这里要特别说一句Flink 自带的 Kafka Source 和 Redis Sink 往往不够用。公司自研埋点 SDK 的上报格式、内部 Redis 集群的鉴权方式都需要自定义 data source 与 data sink这也是 flink 实时计算进阶篇里最常被问到的部分。自定义 Source 的核心是理清消费位点、反压信号和 schema 解析三件事自定义 Sink 的核心则是批量写入、失败重试和幂等。如果公司已经上了 openmetadata 这类数据血缘平台Flink 作业的血缘关系可以被自动采集字段级溯源到 Kafka topic这对后面排查“特征被谁改过”很有价值建议在一开始就把作业名和 topic 命名规范定好血缘平台才能真正发挥作用。3. 从埋点到特征一套最小可复现的实时计算作业看完架构下一步是把用户特征作业落地。下面这套逻辑是实时推荐链路中最基本、也是最先要跑通的部分。它做的事情可以概括为接收行为日志按用户维护最近一段时间的行为序列把这个序列作为后续召回和排序的输入。3.1 行为数据的字段规范决定后续所有计算的上限行为日志字段规范决定特征质量的上限Flink 作业只是把规范变成结果。一条点击事件至少要有这些字段userId、itemId、behaviorTypeclick、cart、order、sceneId首页、详情页、搜索页、ts事件发生时间毫秒时间戳。如果公司埋点已经存在最好在接入层补齐不要指望 Flink 侧做数据清洗字段缺失只能靠默认值硬填后续聚合和 join 会全部偏离。一个常见误区是为了兼容所有业务线把行为日志设计成几十个字段的大宽表。实时推荐真正高频使用的字段通常不超过八个字段越多Kafka 序列化开销和 Flink 反序列化开销越大。我一般处理方式是 topic 里只保留推荐链路必用字段其余字段走旁路给数据仓库这样 Kafka 吞吐和 Flink 作业性能都能保住。3.2 用 KeyedProcessFunction 维护用户实时行为序列下面的代码是用户特征作业的核心骨架。它消费 Kafka 行为日志按 userId 分组用 ListState 保存用户最近 30 分钟的行为序列并用事件时间定时器在 30 分钟后清理状态。DataStreamUserBehavior stream env .addSource(new FlinkKafkaConsumer(user_behavior, new JSONDeserializationSchema(), kafkaProps)) .assignTimestampsAndWatermarks( WatermarkStrategy.UserBehaviorforBoundedOutOfOrderness(Duration.ofSeconds(10)) .withTimestampAssigner((event, timestamp) - event.getTs())) .keyBy(UserBehavior::getUserId) .process(new KeyedProcessFunctionString, UserBehavior, UserRecentItems() { private transient ListStateUserBehavior recentItems; private transient ValueStateLong cleanupTimer; Override public void open(Configuration parameters) { ListStateDescriptorUserBehavior descriptor new ListStateDescriptor(recentItems, UserBehavior.class); recentItems getRuntimeContext().getListState(descriptor); cleanupTimer getRuntimeContext().getState(new ValueStateDescriptor(cleanupTimer, Long.class)); } Override public void processElement(UserBehavior value, Context ctx, CollectorUserRecentItems out) throws Exception { recentItems.add(value); if (cleanupTimer.value() null) { long timerTime ctx.timestamp() 30 * 60 * 1000L; ctx.timerService().registerEventTimeTimer(timerTime); cleanupTimer.update(timerTime); } out.collect(buildOutput(recentItems.get())); } Override public void onTimer(long timestamp, OnTimerContext ctx, CollectorUserRecentItems out) throws Exception { recentItems.clear(); cleanupTimer.clear(); } });逻辑说明这段代码的关键不是 map 和 filter而是状态和时间语义。assignTimestampsAndWatermarks 告诉 Flink 使用事件时间而非处理时间后续窗口和定时器都基于日志里的 ts 推进KeyedProcessFunction 里每个用户维护一份独立的 ListState保存的是行为对象列表而不是拼好的字符串因为后续召回阶段还要从这些对象里取 itemId 和 sceneId定时器注册在第一条事件的事件时间上加 30 分钟时间到达后触发 onTimer 清理状态确保长期不活跃用户的状态不会积压。参数说明forBoundedOutOfOrderness 的 10 秒是乱序容忍度要根据埋点上报链路的实际延迟调整。设得太大窗口出结果变慢特征时效性下降设得太小迟到数据被丢弃特征不完整。30 分钟的行为序列窗口决定“实时偏好”的敏感度做闪购类业务我一般压缩到 5 分钟做常规电商可以放宽到 30 到 60 分钟。ListState 可以额外配置 TTL按天或按小时清理防止序列无限增长。还需要注意如果只写状态而不开 checkpoint进程重启后状态全部丢失实时特征就断了。checkpoint 是实时作业的后悔药建议一上生产就打开。3.3 ItemCF 的实时化物品相似度不能全量算离线 ItemCF 的做法是统计用户行为序列中物品对共现次数再计算余弦相似度。实时化之后最朴素的思路是把全量相似度计算改成“增量共现窗口”只在用户最近点击列表里生成物品对。下面这段代码是核心示意Flink 版本不同 API 命名会有些差异但整体结构是通用的。// 过滤出点击行为按用户分组维护最近 N 个点击物品 DataStreamUserBehavior clicks stream.filter(e - click.equals(e.getBehaviorType())); clicks.keyBy(UserBehavior::getUserId) .process(new KeyedProcessFunctionString, UserBehavior, ItemPair() { private transient ListStateString itemSeq; Override public void open(Configuration parameters) { itemSeq getRuntimeContext() .getListState(new ListStateDescriptor(itemSeq, String.class)); } Override public void processElement(UserBehavior value, Context ctx, CollectorItemPair out) throws Exception { itemSeq.add(value.getItemId()); ListString items new ArrayList(); for (String item : itemSeq.get()) { items.add(item); } // 只保留最近 20 个点击避免列表无限膨胀 if (items.size() 20) { items items.subList(items.size() - 20, items.size()); itemSeq.clear(); itemSeq.addAll(items); } // 把最后一个点击和它之前的所有点击组成共现对 if (items.size() 2) { String last items.get(items.size() - 1); for (String prev : items.subList(0, items.size() - 1)) { out.collect(new ItemPair(prev, last)); } } } });逻辑说明这个算子把每个用户的点击行为按时间顺序累积每次新点击到达时都和当前序列里已有的物品组成共现对。比如用户依次点击了 A、B、C那么在 C 到达时会输出 C-A、C-B 两对。这样做避免了离线方式下全量扫描用户历史的开销只关注“最近发生了关联”的物品。参数说明保留最近 20 个点击是召回质量和计算量的折中。20 太小长尾关联丢失20 太大每个用户每次点击都要生成更多物品对下游共现统计压力成倍增长。生产环境里还可以再加一层处理生成的 ItemPair 进入滑动窗口按对计数比如 30 分钟窗口每 5 分钟滑动一次只保留出现次数超过阈值的物品对写入 Redis这样能过滤掉误触带来的噪声共现。热门物品的相似集合要注意长度控制我一般每个 itemId 最多保留 50 个相似品避免头部商品把长尾商品的曝光全部吃掉。3.4 特征结果落地Redis 和 HBase 的选型与参数特征计算完成之后在线服务要能高效读取。用户行为序列这类“每个用户一条”的数据用 Redis 最合适物品相似度矩阵这类“每个物品多条”的数据也适合放 Redis。QPS 高、数据量在百万级以内Redis 完全扛得住如果用户量上亿、序列特征总量超过 Redis 内存的合理范围就要把序列特征落到 HBase行键设计成 userId 反转加时间戳前缀避免热点分片。写 Redis 时要注意 key 的过期策略。行为序列和物品相似集合都不需要永久保存给 key 设置 TTL 可以防止存储无限膨胀。数据key 模式结构建议 TTL用户最近行为user:recent:{userId}ZSet 按时间戳排序24 小时物品相似集合item:sim:{itemId}Hash 存相似度得分6 小时实时热销榜rank:hot:overallZSet 按曝光/点击加权10 分钟并行度和 Redis 分片也需要对齐。如果写入用的并行子任务数大于 Redis 集群分片数单个分片会过热我一般会在写入前对存储 key 做一次 keyBy 预聚合再直连 Redis。项目初期如果只是照着 flink 菜鸟教程里的词频统计做一遍对理解算子有帮助但真正跑推荐作业时状态、水印、窗口三个概念才是主战场词频统计里 map 加 sink 的三件套支撑不起这套链路的复杂度。4. 实时推荐避坑记录五个翻车现场的根因与解决办法实时推荐链路长每一层都可能出问题。这里整理的是最常见的五个翻车场景每一条都按现象、原因、解决三步写清楚照着排查可以省下大量抓日志的时间。4.1 事件时间与处理时间混用窗口数据整体漂移现象Kafka 里积压了 10 分钟日志后窗口统计出的“最近 5 分钟热门商品”看起来在更新但和页面实际行为对不上热门商品整体滞后。原因作业里有的窗口用处理时间有的用事件时间。处理时间以 Flink 机器本地时钟为准日志在 Kafka 积压后处理时间比事件发生时间晚导致两个窗口的时间基准不一致数据自然漂移。解决全作业统一使用事件时间Kafka Source 配置 WatermarkStrategy上游 topic 的延迟会直接体现在 watermark 里。运维上要关注消费组 Lag 指标Lag 持续上涨说明消费能力不足先解决这个再谈特征准确。排查时可以用 flink 火焰图定位是哪个算子耗时异常先确认瓶颈在 source、窗口计算还是 sink再决定加并行度还是优化序列化。4.2 维表关联报错flink 的 jdbc 连接器异常现象Flink SQL 作业里 join 商品维表运行几个小时后开始报 flink 的 jdbc 连接器异常连接被关闭作业重启后恢复正常过几个小时又复发。原因维表 join 的默认实现是每条数据都查一次数据库Kafka 高峰流量下连接数被打满超时后被连接池回收后续请求拿到失效连接就开始报错。更隐蔽的是连接池回收和 Flink 算子线程之间不同步偶发异常容易被误判为网络问题。解决常见做法是用 lookup join 加缓存。Flink SQL 维表 DDL 里设置 lookup.cache.max-rows 和 lookup.cache.ttl商品维表这类变化不频繁的维度缓存 5 到 10 分钟即可如果流量再大就用 asyncio 改造维表查询或者把维表预加载成广播状态。血泪经验是维表 join 永远不要用同步逐条查询无论单次查询多快都扛不住 Kafka 高峰吞吐。4.3 状态 TTL 设置不当checkpoint 连环失败现象作业运行一段时间后checkpoint 持续超时连续失败触发自动重启用户特征大面积丢失。原因ListState 里存的是完整行为对象每个对象包含多个字段TTL 设置成一天数据量上来后状态体积膨胀checkpoint 序列化的数据量超过默认超时阈值。解决状态量大的作业先评估存储格式优先用紧凑的字节数组而不是完整 JSON 对象字段能省则省可以适当调大 checkpoint 超时时间同时开启增量 checkpoint。状态体积调参有点像玄学本质是把“保留多少历史行为”和“多久能备份完”两件事对齐不是把 Flink 内存参数盲目调大。4.4 Sink 到 Hive 表数据不入表结果静默丢失现象Flink 作业显示正常结束日志里也写了写入成功但 Hive 表查不到数据或者只有最后一批数据。原因常见于流式结果积累为小文件分批写入 Hive并行度大于分区数时多个并行子任务各自提交文件元数据没有合并另外流式写 Hive 如果没开分区提交数据会一直停在 staging 目录flink sink hive表数据不入表就是这么来的。解决Flink SQL 写 Hive 要显式配置分区提交参数指定提交触发时机和文件格式同时控制写 Hive 的并行度不要超过分区数避免小文件碎片。排查时先看 HDFS staging 目录有没有数据有数据说明是提交策略问题没数据说明上游就没输出这时候用火焰图看算子是 idle 还是 busy再决定往上排查 Kafka 消费还是往下排查 sink。最忌讳的是只把 Hive 连接参数改一遍又重跑浪费一个完整窗口周期。4.5 迟到数据被静默丢弃夜间特征缺失现象白天实时推荐表现正常凌晨网络抖动导致部分行为日志延迟到达第二天用户看到的推荐里丢失了昨晚的浏览偏好。原因水印设置了固定的 10 秒乱序容忍度夜里上报链路抖动超过 10 秒后迟到数据被窗口直接丢弃用户特征里缺失了这部分行为。解决给窗口计算增加迟到侧输出用 OutputTag 收集迟到的行为单独处理同时把水印容忍度按业务时段调整。推荐系统对“最近一次行为”的依赖很强宁可在极端场景下多算一次重复统计也不能把迟到数据静默丢进黑匣子里。5. 召回与排序把特征变成用户看到的商品Flink 算出的特征不会直接变成推荐结果中间还要经过召回和排序。这一层的设计决定了实时特征到底能不能转化为页面上的点击率提升。5.1 实时召回候选集的组装一次读取、聚合去重召回的核心是用“用户刚刚的行为”去特征库里取关联物品。在线服务收到推荐请求后先读 Redis 里这个用户的最近行为序列再批量查每个行为物品的相似集合最后合并去重。这里最大的性能隐患是网络往返次数不能用循环逐条请求。// 在线召回服务核心逻辑读取行为序列与相似物品集合 String recentKey user:recent: userId; SetString recentItems jedis.zrevrange(recentKey, 0, 9); ListString candidates new ArrayList(); // 批量获取相似集合避免逐条循环请求 for (String itemId : recentItems) { MapString, String similar jedis.hgetAll(item:sim: itemId); candidates.addAll(similar.keySet()); } // 过滤已购商品按相似度得分排序 candidates.removeAll(purchasedItems);逻辑说明这段代码先在 Redis 里取用户最近 10 个行为物品再逐个取相似集合。生产环境里要把 for 循环改成 pipeline 或 mget减少网络 RTThgetAll 返回的是 Hash 结构的 field-valuefield 是相似物品 IDvalue 是相似度得分。排序时以相似度得分为主排序键行为时间远近为副排序键保证“最近看过的物品的相似品”排在更前面。参数说明zrevrange 取 10 个行为物品是召回覆盖面和查询开销之间的折中每个物品保留 50 个相似品10 个物品就是最多 500 个候选。候选集过大时排序阶段的压力会明显变高TP99 延迟上涨需要配合实际压测来调。这里给出一个常用的初始参数表上线时根据自己的流量调整。参数初始值说明召回宽度10用户行为序列取数量相似集合长度50每个物品保留相似品上限候选集上限500召回去重后的候选总数排序输出条数20最终展示的商品数5.2 轻量排序行为加权和时间衰减的计算方式召回给出候选商品后排序决定最终展示的 20 个。实时链路里排序模型不宜太重常见做法是先做规则加权再叠加一个轻量逻辑回归。规则加权的核心是把行为类型价值差异和时间衰减写清楚下单权重大于加购加购权重大于点击时间越近权重越大。常见的时间衰减公式是 weight 乘以 exp(-lambda * elapsedMinutes)lambda 取值决定衰减速度做秒级时效类业务时 lambda 要调大。排序特征的来源就是 Flink 已经算好的那套结果用户最近 30 分钟的行为序列长度、候选物品的相似度得分、候选物品价格带与用户历史消费价格带的匹配程度、候选物品是否和用户最近浏览品类一致。这些特征拼成向量喂给逻辑回归输出预估点击率。逻辑回归的训练样本可以用曝光日志和点击日志离线构造不需要在线学习训练链路和实时特征链路完全解耦样本延迟一天也不会影响线上推理。5.3 兜底策略冷启动与热销榜的边界新用户没有任何行为序列召回阶段读 Redis 拿不到数据候选集为空。这种情况直接返回热销榜。热销榜可以由同一个 Flink 作业在窗口内统计曝光和点击加权计算写入 Redis 时用独立 key 和更短 TTL避免和个性化特征混在一起。兜底策略还要考虑不同场景的差异未登录用户的兜底应该是全局热销登录但无行为的新用户可以叠加地域和品类偏好搜索后无结果则需要回到搜索前场景的兜底列表。冷启动用户占比较高时实时推荐整体的点击率指标会被明显拉低这是正常现象需要按用户状态分层看指标不要只看全局平均值。6. 上线前怎么验证这套推荐链路真的变好了实时推荐链路很长上线最怕的是直接切流量后指标下跌又找不到原因。验证方式不需要一开始就上实验平台旁路日志法足够解决问题。6.1 旁路日志做离线 AB不切线上流量也能评估在线服务把“实时推荐给出的候选集、排序结果和最终曝光商品”落一份日志再用离线脚本模拟“如果用户当时看到这个结果点击率会是多少”和线上真实曝光点击做对比。这样不切流量也能得到接近 AB 的效果。旁路日志要带上 requestId、用户行为序列、候选集、排序得分和最终曝光商品缺一个字段都没法复盘。日志量比较大时要按天分桶存储方便回溯。6.2 延迟、覆盖率、状态体积三个硬指标验证不能只看点击率。延迟看的是用户点击到特征生效的时延对比 Kafka 消息里的 ts 和 Redis 特征更新时间p95 压到 10 秒以内才算实时覆盖率看有多少比例请求拿到了个性化候选而不是全部落入热销兜底状态体积看 Flink 作业的状态大小和 checkpoint 耗时这两个指标直接决定长稳运行上限。三个硬指标都达标再谈点击率提升才有意义。我自己养成的习惯是每个版本上线前先在测试环境完整跑一套旁路日志连续观察三天趋势再决定要不要推全量。实时推荐链路很长出问题往往不是模型的问题而是埋点、状态、存储和网络里某个环节悄悄变了。这个教训让我现在遇到推荐效果波动时第一反应永远是看 Kafka Lag 和 Redis 慢查询而不是先去调排序模型参数。希望帮到你。本文还有配套的精品资源点击获取