简介：面向大数据与推荐系统方向的学习者，这份基于 Spark 的电影推荐系统设计与实现文档提供论文与配套源码的整合内容，可用于课程设计、毕业设计选题参考或推荐算法实践。内容围绕 Spark、MongoDB 与 Web 技术展开，梳理基于人口统计学、基于内容和协同过滤三类推荐算法的分析与设计，并覆盖用户登录注册、个性化推荐、电影搜索、用户评分及系统测试等实现环节，便于读者理解离线与实时推荐链路。资源包共 1 个 docx 文件，约 7.46MB，目录从绪论、开发技术、系统分析与设计、系统实现到测试逐章展开，既说明三类算法的适用场景，也记录功能与性能测试思路，适合按章节查阅和对照复盘。已有 207 人学习，可作为推荐系统项目搭建、算法选型与论文写作的参考材料。

1. 冷启动那道坎：为什么单一算法一定会出问题

一个只跑协同过滤的电影推荐系统，上线第一天给出的推荐基本等于随机——新用户没有评分，评分矩阵九成以上是空的。这不是调参能解决的，而是大数据场景下绕不开的结构性缺口。基于 Spark 的电影推荐系统通常把内容推荐算法、统计推荐和协同过滤分三层来实现：离线用 Spark SQL 把热门和平均分算成宽表，实时用微批把新行为打成标签，详情页用标签向量算相似度，协同过滤只在评分足够稠密的区间里工作。分层不是架构过度设计，每一层都在替下一层减少一类数据缺陷。搞清楚它为什么必须这样分层，比会调一遍 ALS 的参数更有价值。

2. Spark 的 RDD 与 DAG：内存迭代为什么能把协同过滤从小时级压到分钟级

协同过滤慢，真正慢的地方不是相似度公式本身，而是同一份评分数据被反复搬运。以矩阵分解为例，训练十轮就意味着用户-电影评分矩阵要被按用户重分布十次、按物品重分布十次，每一轮都是一次全量 shuffle 加一次全量聚合。如果这十轮之间每次都把中间结果写入磁盘再读回来，磁盘往返的字节数会是原始数据量的几十倍，算法再优雅也扛不住这种 IO 放大。Spark 之所以成为这类系统的默认选择，核心不在于它“快”，而在于它把迭代过程中的中间结果留在了内存和本地磁盘里，用血缘关系而不是物化副本来表达计算。

2.1 迭代式算法的磁盘账：批处理模型在这里为什么吃亏

传统的批处理模型按 job 组织计算，一个 job 的 map 阶段产出先溢写到本地磁盘，reduce 阶段拉取后再写回分布式存储，下一个 job 从存储里重新读。这套流程在单轮 ETL 里非常稳，但推荐算法天然是多轮迭代：ALS 每轮要固定一方求解另一方，ItemCF 要先算共现再算相似度再截断 TopK，每一轮都依赖上一轮的完整输出。假设评分数据 100GB、迭代 10 轮，每轮落盘加读回就是 2TB 级别的磁盘往返，而且这还没算 shuffle 阶段的溢写。

RDD 的设计动机就藏在这个账里。它把数据抽象成一组不可变、可分区的记录集合，每一次变换（map、filter、join）只记录“怎么从父 RDD 算出来”，不立刻执行。真正触发计算的是行动操作（count、collect、write），此时引擎才把整条变换链编译成 DAG 一次性跑完。中间结果如果需要复用，就显式缓存；不需要复用，就在流水线里直接传给下一个算子，连内存分配都省了。

提示：惰性求值意味着你在代码里写下的每一行 transform 都不会立刻报错，报错点永远在第一个 action 上。排查时把 action 之前的分区数、schema 打出来看，比盯着 transform 代码猜要快得多。

2.2 宽依赖与 shuffle：stage 边界到底切在哪里

DAG 不是一整块，它会被宽依赖切成若干 stage。窄依赖指父分区到子分区是一对一或多对一，比如 map、filter、flatMap、union，这类变换可以在同一个 stage 内流水线执行，数据不落地。宽依赖指子分区的数据来自多个父分区，比如 groupByKey、reduceByKey、join、repartition，这类变换必须把数据按 key 重新分发，也就是 shuffle。

协同过滤里几乎每一个关键步骤都是宽依赖。按用户聚合评分是 groupByKey，用户向量和物品向量做 join 求预测分是 join，把相似度矩阵截断成 TopK 需要全局排序。所以优化 Spark 推荐任务，本质上优化的是 shuffle 的次数和每次 shuffle 的数据量，而不是算子写得漂不漂亮。

// 读取评分表后立刻按 uid 重分区，后续所有以用户为粒度的聚合都复用这个分布 val ratings = spark.read .format("com.mongodb.spark.sql.DefaultSource") .option("uri", "mongodb://127.0.0.1:27017/movie.ratings") .load() .select($"uid", $"mid", $"score", $"timestamp") .repartition(8, $"uid") // 分区数对齐总核数，key 选高基数且分布均匀的 uid .persist(StorageLevel.MEMORY_AND_DISK) // 以电影为粒度统计评分次数，这里必然产生一次新的 shuffle val rateMore = ratings .groupBy($"mid") .agg(count($"score").as("count"))

第一段代码里的repartition(8, $"uid")决定了后续所有按用户聚合的算子能不能省掉一次 shuffle。分区键选 uid 而不是 mid，是因为用户表通常比电影表大一个数量级，按用户分区能让“查某个用户的最近若干条评分”这类操作只扫一个分区。代价是后续按电影聚合时仍要重新分发，这部分成本无法完全消除，只能通过控制分区数减少空 task。

StorageLevel.MEMORY_AND_DISK而不是MEMORY_ONLY，是个血泪经验。评分矩阵在内存里装不下时，MEMORY_ONLY会把整个分区丢掉重算，表现为任务跑到一半突然变慢甚至 OOM；MEMORY_AND_DISK会溢出到本地磁盘，慢一点但不会崩。在只有一台物理机、靠虚拟机模拟集群的环境里，这个选择几乎是必须的。

2.3 persist、checkpoint 与血缘截断：什么时候该剪断链条

缓存解决的是“重复计算”，checkpoint 解决的是“血缘太长”。ALS 迭代二十轮之后，RDD 的血缘链会有几十层，一旦某个 executor 挂掉，Spark 需要沿着这条链从头重算，重算代价可能比整个任务还高。checkpoint 的作用是把当前 RDD 物化到可靠存储，同时切断它的血缘，让恢复点从这里开始。

两者的取舍很明确：persist 留在内存里，快但不可靠，executor 挂了就没了；checkpoint 写到磁盘或分布式存储，慢但能扛故障。常见做法是在迭代开始前 persist 一次评分矩阵，在迭代到第五轮左右 checkpoint 一次，把前面几轮的血缘剪掉。

// 迭代开始前物化，避免每轮都从 MongoDB 重读 spark.sparkContext.setCheckpointDir("hdfs://hadoop102:9000/spark/checkpoint") val base = ratings.checkpoint() // 截断血缘，同时落一份可靠副本 // 迭代训练，每轮只依赖上一轮的因子矩阵 var userFactors = initFactors(numUsers, rank) var itemFactors = initFactors(numItems, rank) for (iter <- 1 to maxIter) { userFactors = solveLeastSquares(base, itemFactors) // 固定物品，解用户 itemFactors = solveLeastSquares(base, userFactors) // 固定用户，解物品 }

checkpoint是惰性触发的，必须有一次 action 才会真正落盘，所以代码里通常紧跟着一个count()或者后续计算。checkpoint 目录要放在所有节点都能访问的位置，伪分布式下用本地路径也能跑，但换到真集群就会因为路径不存在直接失败，这一点在把单机代码搬到多机时非常容易遇到。

2.4 伪分布式集群的资源分配：四个节点怎么分才不互相拖累

这套系统的实际开发环境只有一台物理机，用虚拟机模拟出四个节点：一个 Master 加三个 Worker。这种配置足以验证逻辑正确性，但资源分配稍有不慎就会出现 Master 起来了、Worker 一个都没注册上，或者 executor 之间抢内存导致频繁 Full GC。

# 启动前先确认每个 Worker 的内存配额，默认 1g 在跑 ALS 时不够用 export SPARK_WORKER_MEMORY=4g export SPARK_WORKER_CORES=2 # Master 地址写主机名而不是 localhost，否则 Worker 注册时解析不到 $SPARK_HOME/sbin/start-all.sh

SPARK_WORKER_MEMORY决定单个 Worker 能分配给 executor 的总内存，SPARK_WORKER_CORES决定能并发跑几个 task。四个虚拟节点跑在同一台物理机上，物理内存被四等分，每个 Worker 给 4g 已经接近上限。很多人在这里出问题的表现是：Web UI 上 worker 列表只有一个，或者 executor 反复启动失败，本质都是内存超配导致的资源协商失败。

另一个容易被忽略的参数是spark.sql.shuffle.partitions，默认值是 200。在几十万条评分的小数据集上，200 个分区意味着每个 task 只处理几百条记录，调度开销远大于计算本身，任务时间反而被拉长。把默认值改成 8 或 16，在伪分布式环境下会看到明显差别。

3. 离线统计推荐：Spark SQL 如何把“热门”算成六张可查的宽表

协同过滤负责个性化，但它回答不了“现在什么片子最火”这种群体性问题。基于人口统计学的推荐算法在这里承担兜底角色：它不需要任何单个用户的历史行为，只统计群体行为分布，因此天然免疫冷启动。这份资料里把它落成了四类统计任务，每类任务的输出都是一张可以直接按 key 查的宽表。

3.1 四类统计回答的是四个不同的问题

历史热门统计的是所有评分记录里被评次数最多的电影，反映的是长期累积的关注度。近期热门统计的是近一个月时间窗内的评分次数，反映的是当下的热度，这两个指标在同一部电影上经常给出完全相反的排序——一部经典老片历史评分量极高但近期无人问津，一部新片则相反。

平均评分统计的是每部电影的分数均值，它衡量质量而不是热度。问题在于，一部只有三个人打十分的电影平均分是 10，一部有十万人打八点五分的电影平均分是 8.5，前者在纯均值排序里会碾压后者。这就是评分人数阈值存在的意义，也是这套系统同时保留热门表和平均分表的原因——两种信号必须分开存，在业务层再做加权。

分类 Top10 统计的是每个题材下评分最高的十部电影。它解决的是“我只想看科幻片”这类明确意图的查询，也让推荐结果具备可解释性：用户看到推荐理由写的是“科幻片高分榜”，比看到“猜你喜欢”更容易接受。

注意：平均评分表在写入时如果不过滤最小评分人数，新上映的片子会因为样本太少霸占榜首，这类结果在演示时看着很亮眼，实际业务里是典型的噪声推荐。

3.2 DataFrame 管道：从 MongoDB 读到 MongoDB 写

整个离线统计的计算部分全部用 Spark SQL 完成，原因是统计类操作的表达力需求很低，groupBy 加 agg 就够，用 DataFrame API 比手写 RDD 更不容易出错，而且 Catalyst 优化器会自动做谓词下推和列裁剪。

// 近期热门：只统计最近 30 天内的评分，按电影聚合计数 val recentHot = ratings .filter($"timestamp" >= System.currentTimeMillis() - 30L * 24 * 3600 * 1000) .groupBy($"mid") .agg(count($"score").as("count")) .orderBy($"count".desc) // 写入 MongoDB，mode 用 overwrite 保证每次离线任务全量刷新 recentHot.write .format("com.mongodb.spark.sql.DefaultSource") .option("uri", "mongodb://127.0.0.1:27017/movie.RateMoreRecentlyMovies") .mode("overwrite") .save()

时间窗过滤放在聚合之前，是为了让 Catalyst 把过滤条件下推到数据源。如果先聚合再过滤，Shuffle 的数据量会是前者的数倍，在评分表上千万行的规模下这个差别非常直观。mode("overwrite")的选择也有讲究：统计结果是全量快照，用 append 会导致同一部电影出现多条记录，前端查询时必须再加一层去重，把复杂度推给了查询侧。

选 MongoDB 作为存储而不是关系型数据库，核心原因是电影数据本身的形态。一部电影的题材字段天然是数组，演员、导演是嵌套结构，用文档存一行就是一个完整对象，不需要拆成三张关联表再 join 回来。代价是没有强 schema 约束，写入时字段类型写错不会被拦住，这个问题会在后面排查章节里再展开。

3.3 UDF 切分题材标签与 Top10 的窗口排序

分类 Top10 是四类统计里唯一需要拆行的任务。电影表里的题材字段形如Action|Adventure|Sci-Fi，一个字符串里塞了多个类别，要按类别分组就必须先把它拆成多行。这里用自定义函数做切分，再用 explode 把数组展开。

// 切分函数：按竖线分割并去掉首尾空格，空串直接过滤 val splitGenres = udf { (genres: String) => if (genres == null) Array.empty[String] else genres.split("\\|").map(_.trim).filter(_.nonEmpty) } val genreTop10 = movies .withColumn("genre", explode(splitGenres($"genres"))) // 一行电影变多行（每题材一行） .join(avgScores, Seq("mid")) // 拼上平均分 .withColumn("rank", row_number().over( Window.partitionBy($"genre").orderBy($"avg".desc, $"mid".asc) // 加 mid 做稳定排序 )) .filter($"rank" <= 10) .drop($"rank")

split("\\|")里的双反斜杠不能省，竖线在正则里是或运算符号，单反斜杠会被当成普通字符处理，切出来整个字符串原样返回，表现为“每部电影只有一个题材”，Top10 结果里全是重复项。orderBy里加mid作为第二排序键，是为了让排序稳定——只按平均分排，分数相同的电影在不同次运行里顺序可能不同，测试时会出现“结果每次都不一样”的假故障。

另一种等价实现是先做电影与题材的笛卡尔积再用条件过滤，这在逻辑上更直观，但笛卡尔积会产生 电影数 × 题材数 的中间结果，再过滤掉大部分，代价比 explode 高一个量级。这份资料里的原始设计走的是笛卡尔积匹配那条路，工程实现上换成 explode 更划算，两种写法产出的结果集是一致的。

3.4 六张宽表的本质：把计算从查询路径上搬到离线

系统最终落到存储层的是六张表：用户表、电影平均评分表、分类 Top10 表、历史热门表、近期热门表、用户推荐表。前五张是统计产物，最后一张是协同过滤产物，全部由离线任务定期刷新。

这个设计的核心取舍是把计算成本前移。前端一次“热门推荐”请求如果实时去聚合评分表，在千万级数据上要几秒；查一张已经排好序的宽表，走索引是毫秒级。代价是数据新鲜度——离线任务一小时跑一次，这一小时内的新行为不会反映在热门榜上。对于电影这种生命周期以周为单位的物品，这个延迟完全可以接受；换成新闻或短视频，同样的架构就必须把窗口缩到分钟级，或者干脆改成流式聚合。

Redis 在这个架构里补的正是新鲜度这一环。离线算出的用户推荐列表和相似度矩阵写进 Redis，实时链路的中间结果也写进去，前端查询先打 Redis，miss 了再回落到 MongoDB。这个双写结构在后面排查章节会碰到一个典型问题：key 的过期时间设错，会导致推荐结果永远停在某个历史时刻。

4. 协同过滤与内容推荐：一张稀疏评分矩阵和一份标签向量表的两套数学

统计层解决的是群体信号，个性化必须回到个体。这一层要处理的核心矛盾是：评分矩阵极度稀疏，而相似度计算需要足够多的共现样本。协同过滤和内容推荐实际上是在用两种不同的方式绕开这个矛盾——前者靠矩阵分解把稀疏矩阵压到低维空间，后者干脆不看评分，只看物品自身的属性。

4.1 稀疏度：评分矩阵的问题根源在哪里

一个十万用户、两万部电影的系统，矩阵有二十亿个格子，实际有评分的可能只有几百万，填充率不到百分之一。在这种密度下算用户相似度，两个用户只要共同评过一部电影，余弦相似度就可能等于 1，系统会把这两个人当成完全同好，推荐结果自然离谱。

# 计算评分矩阵的稀疏度，这是判断协同过滤能不能用的第一指标 num_users = ratings.select("uid").distinct().count() num_items = ratings.select("mid").distinct().count() num_ratings = ratings.count() density = num_ratings / (num_users * num_items) print(f"users={num_users}, items={num_items}, density={density:.4%}")

这段统计在项目开始阶段就该跑一次。密度低于千分之一时，UserCF 基本不可用，ItemCF 也需要设置最小共现阈值才能过滤噪声，矩阵分解反而是更稳的选择，因为它不要求用户之间直接共现，而是通过隐因子间接关联。

4.2 ItemCF：共现点积除以模长，再用阈值截断

ItemCF 的思路是“看过 A 的人也看了 B”。实现上先按用户聚合出他评过分的电影集合，在集合内部做两两组合，把评分乘积累加得到共现强度，最后除以两部电影各自的评分模长做归一化。

// 按用户聚合，得到每个用户评过分的 (电影, 分数) 集合 val userItems = ratings .select($"uid", $"mid", $"score") .rdd .map(r => (r.getInt(0), (r.getInt(1), r.getDouble(2)))) .groupByKey() // 用户内部两两组合，i < j 保证每对只算一次，避免重复计数 val coOccur = userItems.flatMap { case (_, items) => val seq = items.toSeq for { a <- seq b <- seq if a._1 < b._1 } yield ((a._1, b._1), a._2 * b._2) // 分子：两部电影评分的点积 }.reduceByKey(_ + _) // 分母：每部电影的评分平方和开方，单独算一次后广播复用 val norms = ratings .map(r => (r.getInt(1), r.getDouble(2) * r.getDouble(2))) .reduceByKey(_ + _) .mapValues(math.sqrt) .collectAsMap() val simMatrix = coOccur .map { case ((i, j), dot) => val denom = norms(i) * norms(j) ((i, j), if (denom == 0) 0.0 else dot / denom) } .filter { case (_, sim) => sim > 0.1 } // 低相似度直接丢弃，压缩矩阵体积

reduceByKey而不是groupByKey是这段代码的关键。groupByKey会把同一个 key 的所有 value 拉到一个 executor 的内存里组装成迭代器，在热门电影上这个列表可能有几十万条，直接 OOM；reduceByKey会在 map 端先做局部聚合，网络传输量小一个数量级。

sim > 0.1这个阈值不是拍脑袋定的，而是用来控制相似度矩阵的体积。两万部电影的理论组合是两亿对，全存下来内存根本放不下，加上阈值后通常能压到千万对以内，正好能塞进 Redis。这个阈值调高会牺牲召回率，调低会撑爆缓存，实际项目里要在离线评估时扫一遍不同阈值下的覆盖率和准确率。

ItemCF 在电影场景比 UserCF 更合适，原因是物品的属性比用户稳定。一部电影的题材、演员、风格不会变，相似度矩阵可以离线算一次用很久；用户的兴趣漂移很快，上周爱看文艺片这周可能只看动作片，UserCF 算出来的邻居关系很快就过期了。

4.3 ALS 矩阵分解：把稀疏矩阵拆成两个低维因子矩阵

矩阵分解换了个思路：不去找相似的用户或物品，而是假设每个用户和每部电影都能用一个长度 k 的隐向量表示，预测评分就是两个向量的点积。训练目标是让预测分和真实分的平方误差最小，加上正则项防止过拟合。

import org.apache.spark.ml.recommendation.ALS val als = new ALS() .setMaxIter(10) // 迭代轮数，每轮都是一次全量 join + 最小二乘 .setRegParam(0.1) // 正则项，评分少的用户靠它压住过拟合 .setRank(50) // 隐因子维度，电影场景常见区间 20~100 .setUserCol("uid") .setItemCol("mid") .setRatingCol("score") .setColdStartStrategy("drop") // 冷启动用户预测出 NaN，必须丢弃否则 RMSE 是 NaN val Array(train, test) = ratings.randomSplit(Array(0.8, 0.2), seed = 42L) val model = als.fit(train) val rmse = new RegressionEvaluator() .setMetricName("rmse") .setLabelCol("score") .setPredictionCol("prediction") .evaluate(model.transform(test))

rank的选择直接决定模型容量。设得太小，隐向量表达不了用户的多兴趣，推荐结果趋同；设得太大，在稀疏数据上会把噪声也拟合进去，测试集 RMSE 反而上升。常见做法是从 20 开始翻倍试，取 RMSE 开始走平的那个点。

coldStartStrategy("drop")是最容易漏的参数。测试集里会有训练集没出现过的用户，ALS 对这类用户的预测值是 NaN，如果不丢弃，RMSE 计算出来直接是 NaN，你会以为模型彻底跑飞了，其实只是评估指标被污染。

randomSplit一定要固定 seed。不固定的话每次运行划分出的训练测试集不同，RMSE 会在一个区间里随机波动，调参时根本分不清是参数生效了还是数据变了，这种情况在调参类问题里很耗时。

4.4 内容推荐：标签切词、TF-IDF 加权与详情页相似度

内容推荐完全不依赖评分数据，只依赖电影自身的属性。一部电影由题材、导演、演员、年份等标签构成，把这些标签做成词袋，用 TF-IDF 加权后算余弦相似度，就能找到“和这部很像”的电影。

import org.apache.spark.ml.feature.{RegexTokenizer, HashingTF, IDF} // 题材字段按竖线切分成词袋 val tokenizer = new RegexTokenizer() .setInputCol("genres") .setOutputCol("tokens") .setPattern("\\|") val hashingTF = new HashingTF() .setInputCol("tokens") .setOutputCol("rawFeatures") .setNumFeatures(1 << 12) // 4096 维哈希空间，题材量级下碰撞概率可接受 val idf = new IDF() .setInputCol("rawFeatures") .setOutputCol("features") .setMinDocFreq(1) // 只出现一次的稀有标签也保留，否则新片类型被抹平 val idfModel = idf.fit(hashingTF.transform(tokenizer.transform(movies))) val movieVectors = idfModel.transform(hashingTF.transform(tokenizer.transform(movies)))

setMinDocFreq(1)是个反直觉的设置。默认值会过滤掉出现在少数文档里的词，逻辑上是对的——稀有词没有区分度。但在电影场景里，一部新上映的片子可能只有它自己带某个题材标签，被过滤掉之后这部片子的向量就退化了，详情页推荐会永远推不出它。对小规模标签体系，保留稀有词的收益大于噪声代价。

详情页推荐和实时推荐用的都是这套向量，区别在于触发时机。详情页是用户点了某部电影，系统拿这部电影的向量去和其他电影算相似度，取 TopN；实时推荐是用户刚评完分，系统拿他最近评过的若干部电影的向量做加权平均，再去找相似的。前者是单点查询，后者是短列表聚合，计算量都不大，可以放进实时链路。

这套方法的软肋在标签质量。人工打的标签准确但不全，机器学习抽的标签全但不准，两者混用是常见做法。在电影领域，题材、导演这类结构化字段基本可靠，而“风格”“氛围”这类描述性标签往往缺失严重，这也是为什么详情页推荐的结果经常集中在同一批标签齐全的电影上。

4.5 实时链路：微批处理加 Redis 读取，为什么不实时算相似度

实时推荐的完整路径是：用户提交评分，行为日志进入消息队列，流式任务按批次拉起，从 Redis 读取该用户最近 K 条评分，与离线算好的电影相似度矩阵做加权，得到候选集排序后写回 Redis 供前端拉取。

关键点在于相似度矩阵是离线算好的，实时链路只做矩阵与向量的乘法，不做相似度计算。原因很直接：两万部电影的相似度矩阵是全量数据，任何一次实时计算都要扫全表，延迟根本压不住。把重计算前移到离线，实时只做轻量级的查表和加权，这是所有工业级推荐系统的共同结构。

批次间隔的设定是这条链路的调参核心。间隔太短，每个批次只有几条记录，调度开销占大头；间隔太长，用户评完分要等半分钟才看到推荐变化，体验上就是“点了没反应”。在伪分布式环境下通常设 5 到 10 秒，因为资源有限，批次太密会和其他任务抢 executor。

Redis 在这里承担两个角色：一是存离线产出的用户推荐列表和相似度矩阵，二是存实时链路算出的最新结果。两套数据的 key 必须分命名空间，否则实时结果会覆盖离线结果，或者反过来，表现出来就是推荐列表在某些用户身上反复跳变。

提示：实时链路的中间结果如果只写 Redis 不做持久化，Redis 重启后用户会短暂看到空推荐。给这类 key 设置合理的过期时间，并准备一份离线兜底列表，是成本最低的容错方案。

5. 避坑与排查：从伪分布式集群到实时链路的五类常见故障

这套系统在单机上跑通和在集群上稳定运行，中间隔着一堆看起来毫无关联、实际都有明确成因的故障。下面五条是从环境到算法再到实时链路的高频问题，每条按现象、原因、解决的顺序展开，遇到同类表现可以直接对照。

5.1 集群与环境类：Worker 注册不上、executor 起不来

现象：执行启动脚本后，Web UI 上 worker 列表只有一个条目，或者三个 worker 显示为 DEAD；提交任务时卡在Waiting for resources不动。

原因通常有三个层次。最外层是主机名解析，Master 地址如果写成 localhost，worker 启动时会用自己的主机名去注册，两者对不上就注册失败。中间层是内存配额，worker 申请的内存小于 master 要求的最小值（默认 1g），协商直接拒绝。最内层是防火墙或端口占用，worker 之间的通信端口被别的进程占了。

解决顺序是先从 worker 日志看第一行报错，Failed to connect to master和Not enough memory是两类完全不同的问题。前者去改spark-env.sh里的SPARK_MASTER_HOST为实际主机名，并确认所有节点的 hosts 文件互相能解析；后者把SPARK_WORKER_MEMORY调到 2g 以上，同时确认物理机剩余内存够分。

5.2 数据与 schema 类：读出来全是 null

现象：从 MongoDB 读评分表，DataFrame 打印出来score列整列是 null，或者 groupBy 之后所有分组都归到一个 key 上。

原因是文档数据库没有强类型约束，同一个字段在不同文档里可能是 int、double、字符串甚至是嵌套对象。Spark 的连接器做 schema 推断时按采样结果定类型，采样里恰好都是 int，后面遇到 double 就转成 null。另一种情况是字段名大小写不一致，一部分文档写uid，一部分写userId，推断出来的 schema 只有其中一个。

解决办法是在读取时显式传 schema，不要依赖推断。用StructType把每个字段的类型写死，类型不匹配的记录在读取阶段就暴露出来，而不是流到聚合阶段变成一堆 null 再回头查。字段名不一致的问题只能在写入侧统一，读取侧无法自动修复。

5.3 算法与评估类：RMSE 常年卡住不动、推荐全是热门

现象：ALS 训练十轮，RMSE 从第一轮到第十轮几乎不变，始终在一左右；或者模型评估指标看着正常，实际推荐结果里全是评分人数最多的那几部。

原因分两类。评估卡住通常是coldStartStrategy没设，NaN 污染了指标计算，也可能是randomSplit没固定种子，每轮评估用的测试集都不一样，指标自然乱跳。推荐趋同则是正则项太小加上 rank 太大，模型在稀疏数据上把热门物品的偏置学得极强，对所有用户都给出相似的预测。

解决上，先固定 split 的 seed，确认 RMSE 是可比的；再设coldStartStrategy("drop")，把 NaN 从评估里剔除。针对趋同，把regParam从 0.1 往上调，同时把rank从 50 往下压，观察推荐的品类覆盖率有没有改善。覆盖率这个指标比 RMSE 更能反映推荐质量，RMSE 降了但覆盖率也降了，通常是过拟合的信号。

5.4 标签处理类：分类 Top10 结果重复或漏项

现象：某题材的 Top10 里出现同一部电影两次，或者某个题材一部电影都没有。

重复的根源在切分。split("\\|")之后没有trim，标签前后带空格，Action和Action被当成两个不同的题材，各自分组各自排序，结果合并时看起来就像重复。漏项的原因通常是 explode 之后没有做空值过滤，题材字段为 null 的电影展开成空数组直接消失；也可能是窗口排序时没有 tie-breaker，排名相同的记录被截断掉一部分。

修复方式是把切分、去空格、过滤空串这三步封在一个 UDF 里，保证展开后的每个题材值都是干净的；窗口函数的orderBy加上主键作为第二排序键，让排名唯一。写完先用 count 校验一遍：展开后的总行数应该等于所有电影的题材数之和，对不上就说明切分有问题。

5.5 实时链路类：用户评完分，推荐列表不更新

现象：前端提交评分成功，数据库里也能查到这条记录，但刷新页面推荐列表纹丝不动。

这条链路上有三个断点都可能出问题。第一是流式任务的批次间隔设得太长，或者批次积压导致处理延迟；第二是 Redis 里存用户最近评分列表的 key 设了过期时间，用户隔一段时间再来，历史记录已经被清空，实时推荐没有输入；第三是流式任务的 checkpoint 目录残留了旧的 offset，重启后从很久之前的位置开始消费，看起来像“不更新”，实际是在补历史数据。

排查顺序是从后往前：先看 Redis 里那个 key 还在不在、内容对不对，再看流式任务的批次监控里numInputRecords是不是零，最后检查 checkpoint 目录的时间戳。确认是 checkpoint 问题的话，清掉目录重新提交任务，同时把消费位置策略改成从最新开始，避免每次重启都重放历史。

6. 设计取舍：混合推荐的分层边界、验证方法与缓存成本

三层推荐各自的能力边界，其实在数据分布上就已经划定了。统计层在评分数据几乎为零时也能出结果，代价是完全不个性化；内容层只要有电影元数据就能工作，代价是推荐结果被标签质量锁死；协同过滤层精度最高，代价是必须有足够稠密的评分。把这三层叠起来不是为了让架构图好看，而是让每一层在另外两层失效的区间里顶上。

验证这套结构是否真的在工作，光看离线 RMSE 是不够的。我一般会做三件事：一是抽样若干用户，把他们推荐列表里的电影按题材分布统计出来，如果所有人的题材分布都高度相似，说明个性化没有生效；二是看新用户的首次推荐命中率，这个指标直接反映冷启动兜底是否接通；三是盯着 Redis 的内存占用曲线，用户推荐表是全量用户乘 TopN，十万用户乘五十条，光这一份数据就能吃掉几个 G，缓存撑爆之后所有查询会同时回落到 MongoDB，数据库瞬间被打穿。

最后这条是我踩过的坑。当时只算了相似度矩阵的体积，忘了用户推荐表才是大头，上线第二天 Redis 内存告警，一批 key 被淘汰，前端表现为随机几个用户的推荐列表变成空。从那以后我每次设计这类系统，都会先把“全量用户 × TopN × 单条记录大小”这笔账在纸上算一遍，再决定是缩 TopN、做分片，还是把冷用户的推荐改成实时查统计表。

这份资料里论文部分的系统模块图和那几张推荐流程图，正好对应代码里的四个入口：离线统计、协同过滤、内容相似、实时打分。对着图看代码，比直接读源码快得多，尤其是分层边界和表结构这两块，图上标得很清楚，值得先过一遍再动手跑。

希望帮到你。

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