企业官网建设流程全解析

简介：围绕Spark构建的电影推荐系统设计与实现文档，面向大数据、推荐系统方向的课程设计与毕业设计学习者，以及需要了解协同过滤落地的开发者，帮助梳理从需求分析、算法选型到系统实现与测试的完整技术路线。压缩包共1个docx文件，约7.46MB，以论文正文为主，包含摘要、目录、绪论、开发技术介绍、系统分析与设计、系统实现、系统测试、结论及参考文献等章节。文档围绕基于人口统计学、基于内容和协同过滤三类推荐算法展开，并结合Spark、MongoDB与Web技术说明推荐系统的架构与业务流程，功能层面涉及用户登录注册、个性化推荐、电影搜索和用户评分等模块。已有207人学习，适合参考推荐算法的方案设计、功能模块划分与测试思路，也可作为整理相关论文章节结构和关键技术表述的辅助材料。

1. 首页那几排“猜你喜欢”怎么算出来的：这个 Spark 电影推荐系统能解决什么

打开任何一个电影站点的首页，第一屏往往不是搜索框，而是几排"猜你喜欢""最近热门""同类佳片"。这些位置背后不是静态配置，而是一条从评分日志一路算到页面卡片的链路。这份基于 Spark 的电影推荐系统毕设资源，把这条链路完整做了一遍：用协同过滤、内容推荐算法和人口统计学统计三套逻辑，在 Spark 上跑离线与实时两路推荐，结果落到 MongoDB，再由 Spring 后端和前端页面展示出来。适合两类人：一是要交一份能跑通、能讲清算法细节的推荐系统毕设；二是想找一个带完整工程骨架的大数据入门项目，把它当模板改成自己的业务。

2. 技术选型与分层架构：Spark 算、Mongo 存、Redis 缓存、ES 检索各管一段

2.1 这套组合为什么能被撑起来

推荐系统最怕两件事：算不动和存不下。评分表一旦上到千万级，单机的groupBy就开始喘，ALS 这种要反复迭代矩阵分解的算法更是每次迭代都要重扫一遍数据。Spark 的核心优势就在这里——中间结果放内存，RDD 的惰性求值让多个map/filter能合并成一次遍历，迭代式计算的耗时才压得下来。论文里对这一点的表述是"内存存储中间数据、减少磁盘 I/O"，翻译成工程语言就是：ALS 迭代 10 轮，磁盘只读写一次，剩下九次都在内存里滚。

存储这边选 MongoDB 而不是 MySQL，理由很实际。电影数据天生是嵌套的——一部片子有多个类型标签、多个主演、多个关键词，关系型数据库要拆三张关联表再做 join，MongoDB 一个文档就装下了。而且推荐结果本身就是"给用户一批电影 ID 列表"，存成数组字段直接吐给前端，省掉一次拼装。

这里有个容易被忽略的细节：为什么搜索要用 Elasticsearch 而不是直接在 MongoDB 里用正则查。电影名搜索是典型的"用户输错一个字也要能搜到"的场景，MongoDB 的正则匹配既走不了索引，也没法做相关度排序；Elasticsearch 的分词加打分机制天然适配这种需求，还能顺手把"按类型筛选""按评分排序"一起做了。Redis 的定位则更偏性能优化——首页那几个榜单是每次刷新都要读的，放在 Mongo 里查一次几百毫秒，放 Redis 里就是几毫秒的事。

2.2 三层结构：一次详情页请求到底走了哪些代码

整个系统按 view 层、business 层、算法层切开。view 层是 AngularJS 写的页面，负责渲染和收集用户操作；business 层是 Spring 提供的 REST 接口，负责鉴权、参数校验、组装数据；算法层是独立的 Spark 作业，跑完把结果写回 MongoDB，business 层只负责读。

以"用户点开某部电影的详情页"为例，链路是这样的：前端发GET /movies/{mid}/similar，Spring 控制器收到请求后先去 Redis 里查有没有这部片子的相似电影缓存，命中就直接返回；没命中就去 MongoDB 的相似度结果集里查，查到后回写 Redis；如果连结果集里也没有（比如这部片子太新，离线作业还没覆盖到），就触发一次基于内容算法的实时计算，把这部电影的类型、主演、导演标签拿去和全量电影做匹配，算完返回并缓存。

这个"缓存 → 结果表 → 实时兜底"的三级顺序是我很推荐的做法，它把响应时间从"取决于算法有多重"变成了"取决于缓存有没有命中"。很多同学写推荐系统喜欢在接口里直接调算法，本地测试数据量小的时候看不出问题，一上真实数据，接口就超时了。

提示：算法层和业务层之间只通过 MongoDB 的集合通信，不要用 RPC 直连。这样 Spark 作业重启、升级都不会影响线上接口，算法挂了页面顶多是不刷新推荐位，不会白屏。

2.3 六张结果表：离线算完往哪写，前端从哪读

原始数据只有两张表：movies（电影元数据）和ratings（用户评分，含 uid、mid、score、timestamp）。所有推荐结果都是这两张表派生的，一共六个集合，各自对应一类推荐位。

写进 MongoDB 的文档结构大致长这样，UserRecs里的recs是一个内嵌数组，前端拿到就能直接渲染，不用二次查询：

{ "uid": 10023, "recs": [ { "mid": 889, "score": 4.82 }, { "mid": 1204, "score": 4.71 }, { "mid": 337, "score": 4.65 } ] }

注意：AverageMovies里一定要同时存评分人数。只按平均分排序，一部被一个人打了 5 分的冷门片会永远霸榜，这是统计类推荐最典型的翻车点。

3. 核心算法落地：统计推荐、内容相似度与协同过滤的三段实现

3.1 统计推荐：四类榜单其实是一次 Spark SQL 批处理

论文里把这一块叫"基于人口统计学的推荐"，听起来很学术，落到代码就是几句聚合 SQL。它的价值在于不依赖用户历史行为的深度挖掘，新用户进来也能看到"最近大家都在看什么"，是缓解冷启动最便宜的手段。

四类榜单在一个作业里跑完。第一类是历史热门，按电影维度数评分条数；第二类是近期热门，加一个时间窗口过滤；第三类是平均分，注意要过滤低频；第四类是分类 Top10。

// 只保留近 30 天内有评分行为的电影，时间戳单位按数据源实际情况换算 val recentRatings = ratings .filter($"timestamp" > (System.currentTimeMillis() / 1000 - 30 * 24 * 3600)) // 历史热门：评分人数降序，取前 100 val rateMoreMovies = ratings .groupBy("mid") .agg(count("uid").as("count")) .orderBy($"count".desc) .limit(100) // 近期热门：同样按条数排，但数据源换成窗口过滤后的表 val rateMoreRecently = recentRatings .groupBy("mid") .agg(count("uid").as("count")) .orderBy($"count".desc) .limit(100) // 平均分：必须带上评分人数，且人数低于阈值的不进榜 val averageMovies = ratings .groupBy("mid") .agg(avg("score").as("avg"), count("score").as("cnt")) .filter($"cnt" >= 20)

参数上有三个地方值得说。时间窗口的 30 天不是拍脑袋定的，它决定了"近期热门"多久换一次血，窗口太短榜单会剧烈抖动，太长就退化成历史热门。评分人数阈值 20 是过滤水军的粗筛，真实业务里这个值要和总用户量挂钩，用户百万级的站可以放到 100 以上。取前 100 是为了给前端留出分页和随机抽样的空间，页面上一般只展示 12 到 20 个。

分类 Top10 的写法稍微绕一点，因为movies.genres存的是"Action|Adventure|Sci-Fi"这种竖线分隔的字符串。论文里用 UDF 切分再笛卡尔积匹配，思路没问题但效率一般，我一般直接用explode加窗口函数：

// 1. 把一行电影炸成多行，每行一个类型标签 val movieGenres = movies .select($"mid", explode(split($"genres", "\\|")).as("genre")) // 2. 关联平均分表，按类型分区、按分数排序打上名次 val ranked = movieGenres .join(averageMovies, Seq("mid")) .withColumn("rank", row_number().over( Window.partitionBy("genre").orderBy($"avg".desc, $"cnt".desc) )) .filter($"rank" <= 10) // 3. 再按类型聚回去，一个类型一条文档 val genresTop = ranked .groupBy("genre") .agg(collect_list(struct($"mid", $"avg", $"rank")).as("movies"))

explode把嵌套字段摊平，row_number在分区内排名，最后collect_list收拢回嵌套结构——三步走完，产出的文档结构和前端要的格式几乎一致。这里排序条件写的是avg desc, cnt desc，先比平均分再比人数，就是为了避免"9.9 分但只有 3 个人评"的片子挤掉"8.7 分有两万人评"的经典片。

3.2 基于内容的推荐：一份相似度，详情页和实时流共用

基于内容算法的核心是"给物品打标签、算标签重合度"。它不依赖其他用户的行为，所以天然扛冷启动——新用户只要点开过一部片，就能立刻推出相似的。在这个系统里它被用了两次：详情页推荐（点了 A 推 B）和实时推荐（刚评完 A 推 B）。

标签来源是电影元数据里的四个字段：类型（genres）、关键词（keywords）、主演（cast）、导演（director）。四个字段的区分度不一样，类型最粗但最稳，导演和主演更精准但容易长尾，所以要给不同权重。

// 四个维度的权重：类型最泛，导演最精准 val GENRES_W = 1.0 val KEYWORDS_W = 0.8 val CAST_W = 0.5 val DIRECTOR_W = 0.7 // 单维度打分：交集大小 / 源标签数量，落在 [0, weight] def tagScore(src: Seq[String], dst: Seq[String], w: Double): Double = { val s = src.filter(_.nonEmpty).toSet val d = dst.filter(_.nonEmpty).toSet if (s.isEmpty || d.isEmpty) 0.0 else w * s.intersect(d).size.toDouble / s.size } // 两条电影记录的综合相似度 def similarity(a: Movie, b: Movie): Double = tagScore(a.genres.split("\\|"), b.genres.split("\\|"), GENRES_W) + tagScore(a.keywords.split("\\|"), b.keywords.split("\\|"), KEYWORDS_W) + tagScore(a.cast.split("\\|"), b.cast.split("\\|"), CAST_W) + tagScore(a.director.split("\\|"), b.director.split("\\|"), DIRECTOR_W)

这段代码有三个设计点。第一，intersect之后除以源标签数量而不是并集数量，是为了避免大热片因为标签多、跟谁都有交集而污染推荐结果；第二，每个维度的分数上限等于它的权重，四个维度全中理论最高 3.0 分，实际工程里设 0.6 到 1.2 作为相似度阈值比较合适；第三，主演字段匹配用的是精确字符串，如果数据源里主演名字有别名或空格差异，要在数据清洗阶段统一，否则这一维永远打 0 分。

真正的相似度计算是一个笛卡尔积，一部电影要和其余所有电影比一遍，N 部电影就是 N² 次比较。数据量上万之后必须下手：我一般先把候选集按类型粗筛一遍（只有类型有交集的才进入精算），能把比较次数压掉一个数量级，然后再算四维分数，取 Top 20 写进MovieRecs。

// 粗筛 + 精算：先按类型分组，只在组内两两比较 val byGenre = movies.map(m => (m.genres.split("\\|").toSeq, m)) // 组内自连接，过滤掉自己和分数不够的 val pairs = byGenre.join(byGenre) .filter { case (g, (a, b)) => a.mid < b.mid } // 只算一次，A-B 和 B-A 同价 .map { case (_, (a, b)) => (a, b, similarity(a, b)) } .filter(_._3 >= 0.6)

实时推荐复用的是同一份相似度数据，只是入口不同。用户提交一条新评分后，Streaming 作业从流里拿到(uid, mid, score)，去MovieRecs里查出这部电影的相似电影列表，按相似度 × 本次评分加权累加，排个序取前 10 推给用户。这样算出来的是"因为你刚给这部片打了高分，所以这几部你可能也喜欢"，比离线推荐的解释性强得多，页面上直接就能写推荐理由。

// 每个微批：拿到评分事件，关联相似度，加权打分 stream.foreachRDD { rdd => val events = rdd.map(r => (r.mid, (r.uid, r.score))) // (mid, (uid, score)) val recs = events.join(similarMovieRDD) // (mid, ((uid,score), similarList)) .flatMap { case (_, ((uid, score), similars)) => similars.map(s => ((uid, s.mid), s.sim * score)) // 相似度 * 评分 = 贡献分 } .reduceByKey(_ + _) .map { case ((uid, mid), s) => (uid, (mid, s)) } .groupByKey() .mapValues(_.toSeq.sortBy(-_._2).take(10)) // 每人取 Top10 // 写入 Redis，键为 uid，值为推荐列表 recs.foreachPartition(saveToRedis) }

这段流处理代码的参数含义：similarMovieRDD是广播出去的相似度数据，规模大时应该做成broadcast变量而不是 join，避免每个微批都 shuffle；take(10)决定实时推荐位的数量，一般比离线推荐少，因为实时结果更新快、用户容忍度低；saveToRedis里要用uid做键并设置过期时间，否则内存会一直涨。

3.3 协同过滤：ALS 训练与 UserRecs 落库

协同过滤是整个系统的算法主角，它看的是"和你口味相似的人还喜欢什么"。实现用 Spark MLlib 的 ALS（交替最小二乘），把用户-电影评分矩阵分解成用户隐向量和电影隐向量，再用两个向量的点积预测未评分项。

import org.apache.spark.mllib.recommendation.{ALS, Rating} // 超参数：隐向量维度、迭代轮数、正则系数 val RANK = 20 val ITERS = 10 val LAMBDA = 0.01 // 低频过滤阈值：评分少于这么多条的用户/电影不参与训练 val MIN_RATINGS = 5 val rawRatings = ratings .select("uid", "mid", "score") .rdd .map(r => Rating(r.getInt(0), r.getInt(1), r.getDouble(2))) .cache() // 过滤低频用户和低频电影，缓解数据稀疏 val userCnt = rawRatings.map(r => (r.user, 1)).reduceByKey(_ + _) .filter(_._2 >= MIN_RATINGS).collectAsMap() val movieCnt = rawRatings.map(r => (r.movie, 1)).reduceByKey(_ + _) .filter(_._2 >= MIN_RATINGS).collectAsMap() val bcUser = spark.sparkContext.broadcast(userCnt) val bcMovie = spark.sparkContext.broadcast(movieCnt) val trainData = rawRatings .filter(r => bcUser.value.contains(r.user) && bcMovie.value.contains(r.movie)) .cache() // 训练模型 val model = ALS.train(trainData, RANK, ITERS, LAMBDA)

三个超参的调法我按经验说一下。RANK是隐向量维度，它决定了模型能表达多复杂的口味结构，20 是个中庸值，数据量小的时候设 10 就够，设大了只会过拟合；ITERS是迭代轮数，ALS 一般 10 到 20 轮就收敛了，再往上加收益很小；LAMBDA是正则项，值越大模型越保守，评分数据稀疏的时候要适当调大，防止某个用户只评了 3 部片就训出一个极端的向量。

训练完要预测。策略是给每个用户找"他没评过分的电影"，用模型算出预测评分，排前 20 个写进UserRecs。

// 用户已评分的电影集合 val userRated = trainData.map(r => (r.user, r.movie)).groupByKey() // 待预测的 (uid, mid) 对：全量电影减去已评分的 val candidates = userRated.flatMap { case (uid, rated) => val ratedSet = rated.toSet bcMovie.value.keys.filterNot(ratedSet.contains).map(mid => (uid, mid)) } // 预测并取 Top20 val predictions = model.predict(candidates) val userRecs = predictions .map(r => (r.user, (r.product, r.rating))) .groupByKey() .mapValues(_.toSeq.sortBy(-_._2).take(20))

candidates这一步是性能瓶颈所在。如果电影有 10 万部、用户有 1 万个，理论上要生成上百亿个候选对，内存直接爆掉。工程上的做法是先用业务规则砍一刀：只给"最近 30 天有活跃行为"的用户做推荐，候选集只从"最近上架的、有热度的"电影里选，规模能压到千分之一。这个限制在论文里通常不写，但真跑起来绕不开。

3.4 把几路推荐合并成一个列表

前端首页只有一个推荐位，但后台有四路结果。合并逻辑不能简单拼接，否则用户会看到同一个类型刷屏。我一般在后端接口层做一个简单的打散：按 4:3:2:1 的比例从协同过滤、内容相似、近期热门、平均分榜单里各取几条，中间插入去重和同类型最多连续两条的限制。这一步不上 Spark，纯 Java 代码在 Spring 里做，因为数据量小、又需要低延迟。

4. 环境搭建与启动链路：从伪分布式 Spark 到 8088 端口的登录页

4.1 一台物理机模拟三节点集群

论文里写得很实在：只有一台物理机，用虚拟机模拟小规模集群。三台 CentOS 机器，主机名分别是hadoop102（Master）、hadoop103和hadoop104（Worker），这是大数据项目里最常见的命名习惯。装 JDK、配环境变量、改 hostname、配 ssh 免密，这四步一个都不能少，尤其是免密登录，start-all.sh脚本靠它去各节点起进程，配不好就是"命令执行了但节点起不来"。

# 1. 三台机器都改主机名（各自执行） hostnamectl set-hostname hadoop102 # 另两台改成 hadoop103 / hadoop104 # 2. 配置 hosts 解析，三台机器内容一致 cat >> /etc/hosts <<'EOF' 192.168.1.102 hadoop102 192.168.1.103 hadoop103 192.168.1.104 hadoop104 EOF # 3. 生成密钥并分发，实现免密（在 hadoop102 上执行） ssh-keygen -t rsa -P '' -f ~/.ssh/id_rsa ssh-copy-id hadoop102 ssh-copy-id hadoop103 ssh-copy-id hadoop104

ssh-copy-id之后一定要手动ssh hadoop103验一次，能免密码登录才算真的通了。这一步翻车率极高，常见的坑是 hosts 里的 IP 和虚拟机实际网段对不上，或者防火墙没关导致端口不通。

接下来配 Spark 自己的环境文件，两个文件：spark-env.sh管运行环境和 Master 地址，slaves管 Worker 列表。

cd $SPARK_HOME/conf cp spark-env.sh.template spark-env.sh cat >> spark-env.sh <<'EOF' export JAVA_HOME=/usr/local/jdk export SPARK_MASTER_HOST=hadoop102 export SPARK_MASTER_PORT=7077 export SPARK_WORKER_MEMORY=2g export SPARK_WORKER_CORES=2 EOF # Worker 节点列表 echo -e "hadoop103\nhadoop104" > slaves

SPARK_MASTER_HOST必须写主机名或内网 IP，写localhost的话 Worker 连不上 Master；SPARK_WORKER_MEMORY按虚拟机内存给，每个 Worker 2G 是跑 ALS 的底线，给太少会频繁 GC 甚至 OOM。配完执行start-all.sh，然后浏览器打开 Web UI 看 Worker 列表里有没有两个节点在线。论文里记录的访问地址是hadoop102:8989，说明spark.ui.port被改过；如果你没改这一项，默认端口是 8080，别照着文档输却不改配置。

4.2 MongoDB、Redis、Elasticsearch 三件套的初始化

MongoDB 装在本机，建一个叫movie的库，把movies和ratings两张原始表导进去。数据量不大时用mongoimport导 CSV 最快：

# 导入电影元数据与评分数据 mongoimport --db movie --collection movies \ --file movies.csv --type csv --headerline mongoimport --db movie --collection ratings \ --file ratings.csv --type csv --headerline # 建索引，后面所有作业都按这两个字段查 mongosh movie --eval ' db.ratings.createIndex({ uid: 1 }); db.ratings.createIndex({ mid: 1 }); db.movies.createIndex({ mid: 1 }, { unique: true }); '

索引这一步别省。ratings表动辄上百万行，没有uid和mid索引，后面每个 Spark 作业读数据都要全表扫，跑一次等半天。

Redis 用来缓存榜单和相似度矩阵，启动后先塞一份热门榜单进去做预热，避免服务刚起来时首页全空。Elasticsearch 负责搜索，建索引时字段映射要写清楚，name和description用text类型并指定中文分词器，genres用keyword类型方便精确过滤：

curl -X PUT "localhost:9200/movies" -H 'Content-Type: application/json' -d' { "mappings": { "properties": { "mid": { "type": "integer" }, "name": { "type": "text", "analyzer": "ik_max_word" }, "description": { "type": "text", "analyzer": "ik_max_word" }, "genres": { "type": "keyword" }, "score": { "type": "double" } } } }'

4.3 后端打包与 8088 端口验证

后端是 Spring + Maven 的标准结构，配置文件里把三个中间件的地址写清楚，然后mvn package打成 jar 直接跑。

# application.yml server: port: 8088 spring: data: mongodb: uri: mongodb://127.0.0.1:27017/movie redis: host: 127.0.0.1 port: 6379 elasticsearch: uris: http://127.0.0.1:9200

# 打包并启动 mvn clean package -DskipTests java -jar target/movie-recommender-1.0.jar

启动完成后浏览器访问localhost:8088，能跳转到登录页说明整条链路打通了。这里有个检查顺序值得记住：先看 Mongo 连没连上（日志里有没有连接池初始化成功），再看 Redis 的 ping 通不通，最后看 ES 的健康状态是不是 yellow 以上。三个中间件任意一个挂掉，登录页可能还能打开，但登录之后首页会是空的，因为推荐数据读不出来。

提示：Spark 作业是独立提交的，不随 Spring 服务启动。顺序上要先跑一遍离线作业把结果表灌满，再启动 Web 服务，否则首页会一片空白，你还以为是前端渲染出了问题。

5. 避坑与排查：集群起不来、推荐为空、中文搜索翻车的现场记录

5.1 集群与端口类：进程起了但页面打不开

现象：执行start-all.sh之后，Master 的 Web UI 能打开，但 Worker 列表是空的，提交作业报Connection refused。

原因：八成是 Worker 节点上的SPARK_MASTER_HOST没配或者配成了localhost，导致 Worker 启动后试图连接自己本地的 7077 端口。也有可能是 hosts 解析不对，Worker 拿到了一个错误的 Master 地址。

解决：在 Worker 机器上ping hadoop102确认解析正确，然后检查spark-env.sh里SPARK_MASTER_HOST是不是写的主机名。改完要重启整个集群，只重启 Worker 不生效，因为 Master 的地址是在启动时读进内存的。

现象二：Web UI 打不开，但进程明明在。

原因：端口冲突或者端口被改过。论文里写的是 8989，默认是 8080，两个数对不上就容易怀疑人生。

解决：netstat -tlnp | grep java看 Spark 实际监听的端口，然后以spark-defaults.conf里的spark.ui.port为准。云主机上还要检查安全组有没有放行对应端口。

5.2 数据与算法类：推荐列表全是空的

现象：页面能登录，但"猜你喜欢"和"热门电影"都是空白，MongoDB 里的UserRecs集合是空的。

原因：最常见的是低频过滤阈值设得太高。MIN_RATINGS = 5意味着评分少于 5 条的用户会被直接过滤掉，如果测试数据只有几十个用户、每人评了三四部片，过滤完就一条不剩了。另一种可能是评分字段类型不一致，CSV 里的score被当成字符串读进来，ALS.train会因为类型不匹配报错或者产出全零向量。

解决：本地调试时把MIN_RATINGS降到 1，先把链路跑通再逐级调高；读数据时显式cast("score as double")。另外检查ratings里有没有uid或mid为空的行，空值会让Rating构造出非法对象，进而让整个 RDD 计算失败但不报明显错误。

现象二：协同过滤跑出来了，但推荐结果高度重复，每个用户拿到的都是同样那几部片。

原因：数据稀疏。用户评分矩阵里非零元素占比太低，ALS 学不出区分度，最后退化成"按电影热度推荐"。

解决：调低RANK（从 20 降到 10）让模型更保守，同时把LAMBDA调大一点抑制过拟合；业务上再加一路基于内容的推荐做兜底，两者混合排序，至少保证推荐结果有差异性。

5.3 检索与联调类：中文搜不出结果

现象：搜索框里输"星际"，一部片都搜不到；输英文片名反而正常。

原因：Elasticsearch 索引的 mapping 里没指定中文分词器，默认的 standard 分词器会把一整句中文切成单个汉字，倒排索引里存的全是单字，搜"星际"这种双字词自然匹配不上。

解决：装ik分词插件，然后在 mapping 里把analyzer设成ik_max_word（索引时细粒度切分）、search_analyzer设成ik_smart（查询时粗粒度）。改完 mapping 要重建索引重新导数据，只改配置不重建是无效的。

现象二：接口返回 200，但数据是空的，日志里也没有异常。

原因：MongoDB 的集合名大小写不一致。Mongo 的集合名区分大小写，代码里写UserRecs，作业写进去的可能是userrecs，两个是不同的集合。

解决：统一用常量管理集合名，别在代码里散落字符串字面量。排查时直接show collections看一眼实际写进去的名字。

6. 二次开发：换掉 ALS、加一路推荐位该怎么改

拿到这套代码之后，最值钱的改造点不是换个页面配色，而是把算法层做成可替换的。现在的结构里，算法作业和写入集合是硬绑定的——ALS 作业直接往UserRecs写，基于内容的作业直接往MovieRecs写。想加一路新算法，就得复制一份作业再改集合名，改几次就乱了。

我一般会先抽一层统一的输出契约：不管什么算法，产出的都是(uid, mid, score, source)四元组，统一写到一个RecCandidates集合里，source字段标记来源算法。这样加新算法只是多一个生产者，业务层读的时候按source分组，排序合并的逻辑一处维护。改造的关键代码如下，核心是在写库前做一次 schema 归一：

// 所有算法作业的统一输出结构 case class RecCandidate(uid: Int, mid: Int, score: Double, source: String) def saveCandidates(rdd: RDD[RecCandidate]): Unit = { val spark = SparkSession.builder().getOrCreate() import spark.implicits._ rdd.toDF() .write .mode("append") // 追加而不是覆盖，多路结果共存 .format("com.mongodb.spark.sql.DefaultSource") .option("uri", "mongodb://127.0.0.1:27017/movie.RecCandidates") .save() }

mode("append")是关键，原来是overwrite，多路算法跑完互相覆盖，最后只剩最后跑的那一路。改成 append 之后要注意清理策略，否则集合会无限膨胀，一般按source+ 日期做定期删除，或者干脆每次全量重算前先 drop 再 append。

第二个改造方向是替换协同过滤的实现。ALS 是矩阵分解，如果数据里有明显的时间衰减特征（用户三个月前喜欢的类型和现在可能完全不同），可以换成基于时间加权的 ItemCF，或者引入隐语义模型之外的图 embedding 方案。替换的时候接口保持不变，只是内部实现换掉，这就是上面那层契约的价值。

// 替换实现时只要保证返回同样的 RDD[RecCandidate] object ItemCFRecommender { def recommend(ratings: RDD[Rating], topN: Int): RDD[RecCandidate] = { // 1. 算电影两两共现次数 val coOccur = ratings.map(r => (r.user, r.movie)).groupByKey() .flatMap { case (_, movies) => val list = movies.toSeq for (a <- list; b <- list if a != b) yield ((a, b), 1) }.reduceByKey(_ + _) // 2. 用共现次数 × 评分 给用户打分 ratings.map(r => (r.movie, (r.user, r.score))) .join(coOccur) // (movie, ((user,score), (a,b), cnt)) .map { case (_, ((uid, score), (_, cand), cnt)) => ((uid, cand), score * cnt) } .reduceByKey(_ + _) .map { case ((uid, mid), s) => RecCandidate(uid, mid, s, "itemcf") } } }

验证方法很简单但必须走：换算法之后不要直接看页面，先在 Spark 里对同一批用户跑一遍新旧两套结果，算一下重合率。重合率在 30% 到 60% 之间说明新算法确实带来了差异又没跑偏；如果接近 100%，八成是新算法根本没生效，读到的是旧集合；如果接近 0，那要怀疑是不是uid、mid的类型对不上导致完全匹配不上。

第三个小技巧：调参的时候别改代码重编译。把RANK、ITERS、LAMBDA、MIN_RATINGS这几个参数抽到配置文件或者命令行参数里，用spark-submit --conf传进去，一次编译跑十几组参数对比离线指标（RMSE 或者 Top-N 命中率），比改一次代码跑一次快得多。

说个血泪教训。我最早做这类项目时，习惯把算法结果直接覆盖写进集合，跑一次看一次页面，觉得没问题就交付了。后来加第二路推荐才发现，两路作业并行跑的时候互相覆盖，页面上永远只有先跑完的那一路，排查了两天才定位到是overwrite模式的问题。从那以后我每次做多路结果合并，都强制走一遍"先写入带source标记的候选集合，再由业务层合并"的流程，再也不敢让算法作业直接写最终结果表了。

这套资源里论文和源码是配套的，建议的阅读顺序是先照第 4 章把环境跑通、看到 8088 的登录页，再回头读第 3 章的算法实现，顺序反了很容易看不懂那些阈值为什么这么定。希望帮到你。

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