基于Spark的电影推荐系统：ALS协同过滤与实时推荐全链路

简介面向计算机专业毕业设计与大数据推荐系统入门者的文档资料围绕基于 Spark 的电影推荐系统展开从课题背景、国内外研究现状到系统分析、设计、实现与测试形成完整闭环适合需要完成同类毕设或想了解个性化推荐系统实际开发流程的读者参考。压缩包约 7.46MB共 1 个 docx 文件内含摘要、五章正文、结论、参考文献与致谢章节层级清晰可按目录快速定位所需模块。内容覆盖 Spark、MongoDB 与 Web 端相关技术并分别讲解基于人口统计学、基于内容与协同过滤三类推荐算法的原理与设计思路实现章节给出用户登录注册、个性化推荐、电影搜索与用户评分等功能测试部分包含功能测试与性能测试。目前已有 207 人学习可作为撰写论文、梳理推荐流程以及理解离线与实时推荐思路的参考。1. 从一份论文加源码说起它到底交付了什么、适合谁推荐系统这类资源包最常见的情况是论文写得像综述、代码只有一个登录页。这份基于 Spark 的电影推荐系统不太一样离线统计、ALS 协同过滤、基于内容的相似推荐、Spark Streaming 实时推荐四条链路都在代码里落到了具体的 MongoDB 集合和具体的作业入口上配套论文从可行性分析一路写到功能测试和性能测试。它适合三类人——要交课程设计但不想只交增删改查的学生、想把 Spark MLlib 从官方示例跑成完整业务闭环的工程师、需要一份「大数据加推荐算法」端到端参考架构的技术负责人。下面按选型理由、算法落地、环境跑通、踩坑排查、效果验证的顺序拆开讲。2. 技术栈拆解Spark、MongoDB、Redis、Elasticsearch 各守哪一段2.1 计算层为什么是 Spark协同过滤是迭代计算不是一次批处理协同过滤的核心运算是反复求解用户-物品评分矩阵的低秩近似。以 ALS 为例每一轮要交替固定用户矩阵求物品矩阵、再反过来典型轮次在 10 到 20 之间。如果每次迭代都落盘I/O 时间会直接压过计算时间模型训练从分钟级变成小时级。Spark 把中间结果放在内存里的 RDD/DataFrame 中迭代阶段不需要反复读写磁盘这是它在这类场景下比传统批处理框架更合适的第一动机也是整套系统选 Spark 而不是别的计算引擎的根本原因。第二个动机是组件齐全。同一个 SparkSession 里Spark SQL 负责统计类推荐MLlib 负责 ALS 训练Spark Streaming 负责实时评分流入三条链路共用一套数据源和一套序列化格式省掉了跨引擎搬数据的功夫。对一份要能跑通的参考实现来说这个「一个引擎覆盖三类任务」的特性比极致性能更重要因为它把环境复杂度压下来了。第三是语言。项目主体用 Scala 写函数式语法配合 DataFrame API统计逻辑比 Java 短很多同时又可以直接调用 JVM 生态里的库。上手成本主要在隐式转换和类型推断这两块写过两三个作业基本就顺了。下面是初始化 SparkSession 并挂上 MongoDB 连接器的写法是全项目所有作业的入口import org.apache.spark.sql.SparkSession object RecommenderApp { def main(args: Array[String]): Unit { val spark SparkSession.builder() .appName(MovieRecommender) .master(local[4]) // 伪分布式阶段先用 local 模式跑通 .config(spark.mongodb.input.uri, mongodb://127.0.0.1:27017/movieRecommender) .config(spark.mongodb.output.uri, mongodb://127.0.0.1:27017/movieRecommender) .config(spark.sql.shuffle.partitions, 8) // 小数据量下调小避免空任务拖时间 .getOrCreate() spark.sparkContext.setLogLevel(WARN) // 日志刷屏会盖住真正的报错 // ... 各推荐任务在这里依次调用 spark.stop() } }master(local[4])表示用本机 4 个线程模拟执行器等集群搭好之后换成spark://node01:7077业务代码一行不用改这是「先本地后集群」的关键。spark.mongodb.input.uri和output.uri两个都要配只配 input 的话写入阶段会报找不到输出配置。spark.sql.shuffle.partitions默认是 200小数据集上会起一堆几乎空转的 task压到 8 能明显缩短作业时间这个参数在演示环境里的收益比任何代码优化都直接。setLogLevel(WARN)不是可选项INFO 级别下 Spark 的日志量足以让真正的异常信息滚出屏幕。2.2 存储分层MongoDB 存结果、Redis 扛实时、Elasticsearch 管检索这套系统的存储不是「用了一个数据库」而是按读写形态分了三个角色。MongoDB 存算法产出的结果集特点是结构随算法变、读多写少、需要按条件筛选Redis 存实时推荐和热门榜这类要求毫秒级返回的键值Elasticsearch 承担电影标题和标签的模糊检索靠倒排索引解决like %关键词%这种在关系库里必然全表扫描的查询。具体的集合划分是这套代码里比较值得抄的部分它把「算法输出」和「业务数据」彻底分开了集合名写入方内容读取代价User注册接口用户 ID、账号、密码摘要低Movie数据初始化电影 ID、标题、类型、导演、演员中Ratings评分接口用户 ID、电影 ID、评分、时间戳高增长最快AverageMovies离线统计作业每部电影的平均分低GenresTopMovies离线统计作业每个类型评分前 10低RateMoreMovies离线统计作业历史评分次数榜低RateMoreRecentlyMovies离线统计作业近一个月评分次数榜低UserRecsALS 作业每个用户的推荐列表中ItemRecs内容相似作业每部电影的相似列表中写入 MongoDB 的标准写法是先把结果集转成 Dataset再交给 Mongo Spark Connectorimport com.mongodb.spark.MongoSpark import org.apache.spark.sql.functions._ // avgMovies: DataFrame(mid, avg) val avgMovies ratings .groupBy(mid) .agg(round(avg(score), 2).as(avg)) MongoSpark.save( avgMovies.write .option(collection, AverageMovies) .mode(overwrite) // 重跑作业时覆盖避免重复数据堆叠 )mode(overwrite)是这里最容易被忽略的参数。用默认的append重跑一次统计作业AverageMovies里就会出现两份同样的mid前端取平均分时拿到哪一条全看运气这类问题排查起来非常折磨人。round(avg(score), 2)保留两位小数是为了前端展示统一、减小存储体积如果后续要做排序比较建议在内存里保留原始精度只在写库时截断。集合名必须和前端查询里写死的名字一致改名要同步改两个地方这一条值得写进项目说明不然接手的人一定会踩。Redis 的定位是缓存而不是主存储。热门榜这类数据变化频率低、读取频率高放在 MongoDB 里每次请求查一遍演示时看不出问题并发一上来立刻成为瓶颈。常见做法是离线作业算完顺手写一份到 Redis用ZSET存榜单、score 存热度值取 Top N 用ZREVRANGE一条命令搞定。要注意缓存和库里数据的一致性窗口离线作业重跑之后必须主动删一次对应 key否则用户会看到上一版榜单。Elasticsearch 那边要注意索引结构要和电影元数据对齐。标题用text类型并配置分词类型和导演用keyword这样既能做全文模糊匹配又能做精确过滤。建索引和灌数据是两个独立步骤灌之前先确认 mapping 已生效否则动态映射会把本该是keyword的字段推断成text后续聚合和排序都会受影响。2.3 前后端骨架Spring 加 Maven 的分层与响应式前端的数据契约后端是典型的 Controller-Service-Dao 三段式Spring 的 IOC 负责把 MongoDB 客户端、Redis 客户端、Elasticsearch 客户端注入到 Service 里AOP 用来做接口耗时统计和统一异常处理。Maven 的价值在于把 Spark、MongoDB 驱动、Redis 客户端这些依赖的版本冲突挡在编译期不然光序列化库的版本打架就能耗掉半天。依赖的关键几项如下dependencies !-- Spark 以 provided 引入集群上已有运行时打进 fat jar 会让包体积翻几倍 -- dependency groupIdorg.apache.spark/groupId artifactIdspark-core_2.11/artifactId version2.4.4/version scopeprovided/scope /dependency dependency groupIdorg.apache.spark/groupId artifactIdspark-sql_2.11/artifactId version2.4.4/version /dependency dependency groupIdorg.apache.spark/groupId artifactIdspark-mllib_2.11/artifactId version2.4.4/version /dependency !-- Mongo Spark Connector 必须和 Scala 二进制版本对齐 -- dependency groupIdorg.mongodb.spark/groupId artifactIdmongo-spark-connector_2.11/artifactId version2.4.2/version /dependency /dependenciesscope设为provided的那几项意思是编译时需要、运行时不打包因为集群的SPARK_HOME/jars目录下已经有了。漏了这个 scope打出来的 fat jar 会带上一份 Spark 核心包提交作业时出现类加载冲突报错信息通常是NoSuchMethodError或者ClassNotFoundException指向的类名看起来完全无关这是最容易走弯路的一类问题。另外_2.11这个后缀必须和集群的 Scala 版本一致2.11 的包放到 2.12 的集群上跑编译能过运行必挂。前端用响应式框架页面要适配 PC 和移动端两种宽度。数据契约上有个细节值得提前定死后端返回的推荐列表统一为{mid, title, score, reason}四字段结构reason标这条推荐来自哪个算法热门、相似、协同前端只负责按reason分组渲染。好处是后面加算法不用改前端坏处是reason的取值必须两边对齐建议用一个枚举类在后端集中管理、前端从接口拉取别在代码里散落字符串字面量。2.4 版本与端口规划先定表再动手版本不一致是这套技术栈最常见的翻车点建议动手之前先把这张表定下来全组统一组件建议版本端口说明JDK1.8-Spark 2.4 只稳定支持到 8Scala2.11.x-必须与 Spark 依赖后缀一致Spark2.4.x7077 / 8080Master 端口 / Web UIMongoDB4.x27017单机即可副本集非必需Redis5.x6379只做缓存可关闭持久化Elasticsearch7.x9200单节点关闭集群发现后端服务-8088前端访问入口把 JDK 定在 1.8 不是保守是 Spark 2.4 在更高版本 JDK 上跑会遇到模块系统的反射限制报InaccessibleObjectException网上能搜到的解决方案大多基于更新版本对本项目不适用。这一条列进表格能省掉很多试错时间。3. 三类算法落地从统计榜到 ALS 矩阵分解的完整链路3.1 人口统计学路线四个统计作业撑起冷启动兜底基于人口统计学的推荐在这个项目里承担「非个性化兜底」角色。新用户没有行为数据协同过滤给不出结果这时候热门榜就是唯一能拿得出手的东西。代码里把它拆成四个统计作业每个对应一个 MongoDB 集合。历史热门按电影统计被评分次数取前 Nval rateMoreMovies ratings .groupBy(mid) .agg(count(mid).as(count)) .orderBy(desc(count)) .limit(50) MongoSpark.save(rateMoreMovies.write .option(collection, RateMoreMovies) .mode(overwrite))近期热门要加时间窗口只统计近 30 天的评分。直接用毫秒时间戳做范围过滤不好读一般会先把它压成月粒度的整数import java.text.SimpleDateFormat import java.util.{Date, TimeZone} // 把 13 位毫秒时间戳转成 yyyyMM 的整数方便做范围过滤 val getMonth udf((ts: Long) { val fmt new SimpleDateFormat(yyyyMM) fmt.setTimeZone(TimeZone.getTimeZone(Asia/Shanghai)) // 显式时区别依赖 JVM 默认 fmt.format(new Date(ts)).toInt }) val recent ratings .withColumn(month, getMonth(col(timestamp))) .filter(col(month) 202401) // 换成当前月份往前推一个月 .groupBy(mid) .agg(count(mid).as(count)) .orderBy(desc(count)) .limit(50)UDF 在这里的作用是把时间戳压成月粒度比直接比较毫秒数好读也比字符串比较快。参数上唯一的坑是时区SimpleDateFormat默认取 JVM 时区集群节点时区不一致时会算错边界月份。稳妥做法是像上面这样显式指定或者干脆在数据入库时就把月份字段打好让计算阶段只做过滤。UDF 本身有序列化开销能改写成内置函数就改from_unixtime加date_format的组合可以替代大部分场景。平均分统计更简单但要处理「评分人数太少导致平均分虚高」的问题。一部电影只有一个人打了 5 分平均分就是 5排榜首毫无意义。工程上常见的补救是加最小评分人数门槛比如count 20才进榜更严谨一点用贝叶斯平均把全局平均分作为先验项混进去公式是(v/(vm))*R (m/(vm))*C其中 v 是该电影评分人数、m 是门槛值、R 是电影均分、C 是全局均分。分类 Top10 是全套统计里最容易写错的一段。电影的genres字段是动作|科幻|冒险这种竖线分隔的字符串要按类型分组取前 10得先炸开再分组val splitGenres udf((s: String) s.split(\\|)) val genreTop movies .withColumn(genre, explode(splitGenres(col(genres)))) .join(avgMovies, Seq(mid)) // 带上平均分 .withColumn(rank, row_number() .over(Window.partitionBy(genre).orderBy(desc(avg)))) .filter(col(rank) 10)explode把一行炸成多行row_number配合partitionBy做组内排名这两个组合是 Spark SQL 里做 Top N 的标准套路。rank 10的过滤必须放在窗口函数之后不能提前过滤否则排名会错位。数据量再大一些的话explode之后行数会膨胀好几倍可以在炸开之前先过滤掉评分人数不足的电影减少 shuffle 的数据量。3.2 基于内容的路线把类型、导演、演员组成可比较的特征向量内容推荐的输入是电影自身的元数据输出是「和这部电影相似的其他电影」用在详情页的相似推荐位置。它不依赖用户行为对冷启动友好但效果上限取决于标签质量。这也是「内容推荐算法」这几个字在实际代码里最具体的落点。特征构造常见的做法是字段分桶加权类型权重最高0.5导演次之0.3主演0.2。理由很直白类型决定观影预期导演决定风格演员的号召力在跨类型时不稳定。把每个字段拆成词、拼成一个稀疏向量、权重按字段乘上去然后算余弦相似度。import org.apache.spark.ml.feature.HashingTF // 把元数据拼成一个文档重复字段靠重复词体现权重 val docs movies .withColumn(doc, concat_ws( , repeat_join(col(genres), 5), // 类型重复 5 次 repeat_join(col(director), 3), // 导演重复 3 次 col(actors) // 演员 1 次 )) .select(mid, doc) // HashingTF 把词映射成固定维度向量省掉词表维护 val hashingTF new HashingTF() .setInputCol(doc) .setOutputCol(rawFeatures) .setNumFeatures(4096) // 2 的幂哈希碰撞少且位运算快 val featurized hashingTF.transform(docs)setNumFeatures(4096)这个参数要看数据规模定电影数量在几千到几万级别时4096 维够用再大就会出现明显哈希碰撞把不相关的词映射到同一维相似度算出来全是噪声。判断办法是抽样看几部明显不同类型的电影如果它们的余弦相似度超过 0.6说明维度不够。repeat_join是自定义函数作用是按次数把字段值重复拼接用重复次数代替权重乘法这样后面直接用词频即可不需要再做一次向量乘法。相似度计算有两条路子小数据量直接暴力两两算mid数量上万时开销涨得很快可以改用分布式矩阵的列相似度计算它内部用抽样和分块降低计算量代价是精度有损。演示场景下暴力算完全够几千部电影的相似矩阵几秒钟就出来了。// 暴力两两余弦n 部电影是 n^2 次比较 val vectors featurized.select(mid, rawFeatures).rdd.map { row (row.getAs[Int](mid), row.getAs[org.apache.spark.mllib.linalg.Vector](rawFeatures)) } vectors.cartesian(vectors) .filter { case ((idA, _), (idB, _)) idA ! idB } .map { case ((idA, vA), (idB, vB)) (idA, idB, cosine(vA, vB)) } .filter(_._3 0.3) // 相似度阈值低于它的不写库 .groupByKey() .mapValues(_.toSeq.sortBy(-_._3).take(10))cartesian是笛卡尔积内存和计算量都是平方级必须在中间结果上及时filter不然 shuffle 阶段直接 OOM。0.3这个阈值是经验值太低会把毫不相干的电影塞进列表太高则小众电影凑不满 10 条实际调的时候看一眼结果列表的观感比看任何指标都快。take(10)之前一定要sortBygroupByKey不保证顺序。3.3 协同过滤路线ALS 隐语义模型训练与推荐生成ALS 是这套系统里唯一需要训练的模型也是效果上限最高的部分。它的假设是每个用户和每部电影都能用一组低维隐向量描述评分近似等于用户向量和物品向量的点积。训练过程交替固定一侧优化另一侧所以叫交替最小二乘。import org.apache.spark.ml.recommendation.ALS import org.apache.spark.ml.evaluation.RegressionEvaluator // 只用有效评分0 分或缺失分不参与训练 val cleanRatings ratings .select(col(uid).as(user), col(mid).as(item), col(score).as(rating)) .filter(col(rating) 0) val Array(training, test) cleanRatings.randomSplit(Array(0.8, 0.2), seed 42L) val als new ALS() .setUserCol(user) .setItemCol(item) .setRatingCol(rating) .setRank(20) // 隐向量维度 .setMaxIter(15) // 迭代轮数 .setRegParam(0.1) // 正则防过拟合 .setNonnegative(true) // 评分非负约束后解释性更好 .setColdStartStrategy(drop) // 预测阶段丢弃冷启动用户否则得到 NaN val model als.fit(training) val preds model.transform(test) val rmse new RegressionEvaluator() .setMetricName(rmse) .setLabelCol(rating) .setPredictionCol(prediction) .evaluate(preds) println(sRMSE $rmse)setRank(20)控制隐空间维度值越大表达力越强但越容易记住噪声20 到 50 是协同过滤在中等规模数据上的常见区间。setRegParam(0.1)是正则系数数据越稀疏越需要加大不然用户向量会被个别评分带偏。setNonnegative(true)是对评分场景的合理约束电影评分没有负数加了这个约束后模型不会产出负的预测分省掉一层后处理。setColdStartStrategy(drop)是必设项缺了它评估时遇到训练集里没出现过的用户会返回 NaNRMSE 直接算成 NaN看起来像模型训崩了实际只是少了一个参数。训练完生成推荐列表用recommendForAllUsers一次性给所有用户出结果比循环单用户快得多val userRecs model.recommendForAllUsers(20) // 每个用户 20 条 .select(col(user).as(uid), col(recommendations).as(recs)) .withColumn(recs, explode(col(recs))) .select(col(uid), col(recs.item).as(mid), col(recs.rating).as(score)) // 剔除用户已经评过分的电影 .join(ratings.select(uid, mid), Seq(uid, mid), left_anti) MongoSpark.save(userRecs.write .option(collection, UserRecs) .mode(overwrite))explode把recommendations这个数组字段炸成一行一条方便按uid查询也方便后续过滤。left_anti这个 join 类型值得记一下它的语义是「左表里在右表找不到匹配的行」正好用来剔除已评分电影比filter(not(col(...).isin(...)))高效得多也不用把用户看过的电影列表拉到驱动端。3.4 实时链路Spark Streaming 把新评分折进推荐列表实时推荐的触发点是用户提交评分。评分写入 MongoDB 之后同时把一条消息打到消息队列流式作业消费这条消息更新该用户的实时推荐并写 Redis。这里的「实时」不是重新训练模型而是用已有的物品相似度和用户近期行为做一次快速重排真按流式重训 ALS 成本撑不住也没必要。import org.apache.spark.streaming.{Seconds, StreamingContext} val ssc new StreamingContext(spark.sparkContext, Seconds(5)) // 演示阶段用 socket 模拟消息源生产环境换成消息队列直连 val lines ssc.socketTextStream(node01, 9999) lines .map(_.split(,)) .filter(_.length 3) .map(arr (arr(0).toInt, arr(1).toInt, arr(2).toDouble)) // uid, mid, score .foreachRDD { rdd rdd.foreachPartition { partition val jedis RedisPool.get() // 从连接池取别在循环里新建 partition.foreach { case (uid, mid, score) val sims jedis.zrevrange(ssim:$mid, 0, 9) // 取该电影的相似列表 sims.foreach { case (simMid, simScore) jedis.zincrby(srealtime:$uid, simScore * score, simMid) } } RedisPool.recycle(jedis) } } ssc.start() ssc.awaitTermination()Seconds(5)是批间隔演示时设小一点能看到效果实际部署按数据量调太小会让每批的任务调度开销占比过高。foreachPartition而不是foreach是必须的前者每个分区建一次连接后者每条记录建一次连接开销会直接压垮 Redis。RedisPool要自己实现核心是连接池加try-finally回收没回收的连接会泄漏跑一段时间后 Redis 拒绝新连接报Could not get a resource from the pool看起来像网络问题实际是池子被占满了。写 Redis 的zincrby是累加操作同一部电影被多次加权会叠加分数这正好符合「越相似的越靠前」的直觉。但要注意它不是幂等的同一批评分重放会导致分数虚高做故障恢复时要么保证消息不重放要么在 key 上加批次标记做隔离。4. 环境跑通三节点伪分布式到 localhost:8088 登录页4.1 主机规划、免密登录与 Spark 集群配置只有一台物理机的情况下用虚拟化软件开三台虚拟机模拟小集群是最经济的方案一个 Master 加两个 Worker主机名分别设为 node01、node02、node03每台分 2 核 4G磁盘 40G 起步。这个规格跑几千部电影、几万条评分的数据集完全够再往上就得考虑加内存了ALS 训练阶段的内存峰值比想象中要吃资源。基础环境三件事装 JDK、配环境变量、配免密。前两件是常规操作重点是免密Master 要能直接 ssh 到两个 Worker否则启动脚本会卡在密码输入上# 三台机器都执行生成密钥对并分发公钥 ssh-keygen -t rsa -N -f ~/.ssh/id_rsa ssh-copy-id node02 ssh-copy-id node03 # 验证不该提示输入密码 ssh node02 hostname # 三台机器的 /etc/hosts 都要有同样的映射 # 192.168.56.101 node01 # 192.168.56.102 node02 # 192.168.56.103 node03-N 表示不设密码短语脚本自动分发时必须有这个不然每次要交互。三台机器的/etc/hosts必须保持一致只改一台会出现「Master 能连 Worker、Worker 找不到 Master」的单向故障症状是 Worker 注册不上Web UI 里看不到节点。主机名用hostnamectl set-hostname改完之后要重启网络服务只改文件不重启是无效的。Spark 的配置集中在conf/spark-env.sh和conf/workers两个文件里# conf/spark-env.sh export JAVA_HOME/usr/local/jdk1.8 export SPARK_MASTER_HOSTnode01 export SPARK_MASTER_PORT7077 export SPARK_WORKER_MEMORY3g export SPARK_WORKER_CORES2 # conf/workers node02 node03SPARK_WORKER_MEMORY3g是留给执行器的内存虚拟机总共 4G 的话别设满系统和 Master 进程还要占。设得太满的表现是 Worker 启动正常但一提交作业就有节点失联日志里能看到被系统杀掉。SPARK_MASTER_HOST写主机名而不是 IP换环境的时候只改 hosts 就行少改一处少一个坑。启动和验证$SPARK_HOME/sbin/start-all.sh # 启动 Master 和所有 Worker jps # 每台机器上确认有 Master 或 Worker 进程 # 浏览器访问 http://node01:8080Workers 列表里应有两个 Alive 节点jps是最快的健康检查手段Master 节点上应该看到MasterWorker 节点上应该看到Worker。start-all.sh这个脚本在新版本里被标记为弃用用start-master.sh加start-workers.sh分开执行更稳报错信息也更容易定位到具体是哪一侧没起来。4.2 数据初始化把电影、评分、用户灌进 MongoDB集群起来之后第一步是把数据灌进去。目录结构一般是Movie、Ratings、User三份原始数据格式可能是 JSON 或者分隔符文本。注意Ratings的字段顺序和类型用户 ID 和电影 ID 用整型还是字符串必须一次定死中途改类型会导致所有 join 失效。灌数据用命令行工具最快# 导入电影元数据--jsonArray 表示文件是一个 JSON 数组 mongoimport --host 127.0.0.1 --port 27017 \ --db movieRecommender --collection Movie \ --file movies.json --jsonArray --drop # 导入评分数据一行一个 JSON 对象 mongoimport --host 127.0.0.1 --port 27017 \ --db movieRecommender --collection Ratings \ --file ratings.json --drop # 建索引uid、mid 上的查询和 join 全靠它 mongosh movieRecommender --eval db.Ratings.createIndex({uid: 1}); db.Ratings.createIndex({mid: 1}); db.Movie.createIndex({mid: 1}, {unique: true}); --drop表示导入前先删同名集合重复导入时用它避免数据翻倍正式环境慎用。索引这一步很多人会跳过觉得数据量小无所谓但 ALS 训练前要多次按mid过滤没有索引时每次都是全集合扫描训练时间能差出好几倍。mid上的唯一索引还有个副作用是把脏数据挡在门外重复的电影记录会导致关联时出现一对多结果集行数莫名变多。4.3 启动顺序与三处验证整套系统涉及的服务多启动顺序错了会出现「服务起来了但接口 500」的情况。稳妥的顺序是MongoDB → Redis → Elasticsearch → Spark 集群 → 离线作业 → 后端服务。前四个是基础设施离线作业依赖它们全部就绪后端服务依赖离线作业产出的集合。# 1. 基础设施 mongod --config /etc/mongod.conf --fork redis-server /etc/redis.conf --daemonize yes $ES_HOME/bin/elasticsearch -d # 2. Spark 集群 $SPARK_HOME/sbin/start-master.sh $SPARK_HOME/sbin/start-workers.sh # 3. 离线作业按依赖顺序执行 spark-submit --class com.rec.StatJob --master spark://node01:7077 rec.jar spark-submit --class com.rec.ContentJob --master spark://node01:7077 rec.jar spark-submit --class com.rec.ALSJob --master spark://node01:7077 rec.jar # 4. 后端服务 java -jar movie-web.jar --spring.profiles.activeprod三个验证点缺一不可。第一处是 Spark Web UIhttp://node01:8080里能看到两个 Alive 状态的 Worker说明集群本身没问题。第二处是 MongoDB用db.AverageMovies.count()确认集合里有数据空集合说明作业跑失败了这时候要回去看spark-submit的控制台输出而不是继续往下走。第三处是浏览器访问http://localhost:8088能看到登录页说明后端和前端通了注册一个账号、搜一部电影、点一次评分才算真的跑通。我一般会把这个顺序写成一个 shell 脚本每个步骤本文还有配套的精品资源点击获取