简介这份资源面向推荐系统入门与进阶开发者提供一套基于Spark MLlib实现的豆瓣电影推荐系统完整项目帮助理解协同过滤在真实场景中的落地方式。项目以ALS算法为核心涵盖数据预处理、训练测试集划分、参数调优、评分预测与RMSE、MAE等指标评估并涉及覆盖率与多样性等推荐质量维度适合作为课程设计、毕业项目或大数据分析练手素材。压缩包共4个文件约6.23MB包含pom.xml依赖配置、Scala源码、Shell提交脚本及数据压缩包覆盖从工程构建到集群提交的完整链路目录结构清晰便于按模块阅读与二次开发。目前已有595人学习下载可帮助读者快速掌握Spark MLlib协同过滤的建模流程与优化思路积累大数据推荐系统的实战经验。1. 豆瓣电影推荐系统从评分数据到 Spark ML 的工程化落地豆瓣电影推荐系统这个题目很多人在课程设计或毕业设计阶段都接触过。它表面上是“给用户推电影”实际要解决的是三个工程问题如何把用户对电影的评分行为变成模型能吃的特征、如何用 Spark ML 在分布式环境下完成训练与预测、以及如何把离线训练好的模型接回线上做 Top-N 推荐。我见过太多项目卡在第二步——本地用 pandas 跑通协同过滤一上 Spark 就报内存溢出或序列化错误。这篇文章不讲推荐系统发展史只讲一条能跑通的路径从豆瓣评分数据出发用 Spark ML 的 ALS 算法完成电影推荐并给出参数调优和线上服务的具体做法。适合有 Scala 或 Python 基础、想把这个方向做成可演示系统的从业者。2. 数据准备与特征工程豆瓣评分数据怎么变成 ALS 的输入2.1 豆瓣评分数据的典型结构与清洗边界豆瓣电影评分数据通常包含用户 ID、电影 ID、评分值和时间戳四个核心字段。原始数据常见两种形态一种是爬虫抓取的 JSON 行每个用户一个文件另一种是已经整理好的 CSV每行一条评分记录。无论哪种ALS 只认 (user, item, rating) 三元组所以清洗的目标就是把这四个字段对齐成数值型。常见做法是先把用户 ID 和电影 ID 做整数映射。Spark ML 的 ALS 要求 user 和 item 列必须是整数索引不能直接传字符串。我一般用StringIndexer做映射但要注意StringIndexer默认按出现频率降序编码如果训练集和测试集分开做索引同一部电影可能得到不同编号。正确做法是在全量数据上 fit 一次StringIndexer然后 transform 训练集和测试集。from pyspark.ml.feature import StringIndexer from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(DoubanMovieRec) \ .config(spark.sql.shuffle.partitions, 200) \ .getOrCreate() # 假设原始数据已经读成 DataFrame列名为 user_id, movie_id, rating, timestamp raw spark.read.csv(hdfs:///data/douban/ratings.csv, headerTrue, inferSchemaTrue) # 全量数据上 fit 索引器保证训练/测试编码一致 user_indexer StringIndexer(inputColuser_id, outputColuser_idx).fit(raw) movie_indexer StringIndexer(inputColmovie_id, outputColmovie_idx).fit(raw) indexed user_indexer.transform(raw) indexed movie_indexer.transform(indexed) # ALS 要求 rating 是 float 类型 indexed indexed.withColumn(rating, indexed[rating].cast(float)) indexed.select(user_idx, movie_idx, rating).show(5)这段代码的关键在fit的位置。如果先randomSplit再分别fit测试集里出现训练集没见过的电影时StringIndexer会直接抛异常。全量 fit 的代价是索引器看到了测试集信息严格来说有轻微数据泄露但在推荐系统里用户和电影的 ID 空间是固定的这种泄露对评分预测的影响可以忽略。如果在意可以只对训练集 fit然后对测试集用handleInvalidskip跳过未知 ID。参数方面spark.sql.shuffle.partitions默认是 200小数据集上反而拖慢速度。我一般按数据量调百万级评分设 100 到 200千万级设 500 以上。这个参数影响的是 shuffle 阶段的并行度设太小会 OOM设太大任务调度开销高。2.2 评分矩阵的稀疏性与冷启动处理豆瓣评分数据的稀疏度通常在 99% 以上也就是说绝大多数用户只评过几十部电影而电影总数可能上万。这种稀疏性直接决定了 ALS 的隐因子维度不能设太高。我试过在百万级评分上把 rank 设到 200结果训练时间翻了三倍RMSE 反而上升了 0.02。血泪经验是rank 从 10 开始试每次翻倍观察验证集 RMSE 的拐点。冷启动分两种新用户没有评分记录新电影没有被人评过。ALS 对这两种情况都无能为力因为它的预测完全依赖交互矩阵。工程上的补救措施是在推荐结果里混入热门电影。具体做法是用groupBy(movie_idx).count()统计每部电影的评分人数取 Top 100 作为兜底列表。当 ALS 对某个用户返回空结果时直接推热门榜。# 统计电影热度作为冷启动兜底 popular indexed.groupBy(movie_idx).count() \ .withColumnRenamed(count, rating_count) \ .orderBy(rating_count, ascendingFalse) \ .limit(100) # 后续推荐服务里如果 ALS 结果为空就从 popular 里取 popular.show(10)这里有个容易翻车的点热度统计要用全量数据不能只用训练集。因为冷启动兜底是给线上服务的线上遇到的新用户可能对应任何电影用全量数据统计更稳。但如果你要做离线评估评估集里的热门电影不能来自测试集本身否则指标会虚高。我一般准备两份热度表一份全量用于线上一份仅训练集用于离线评估。3. Spark ML 的 ALS 训练参数怎么设、模型怎么存3.1 ALS 的四个核心参数与调参顺序Spark ML 的ALS有四个参数直接决定推荐质量rank、maxIter、regParam、alpha。rank是隐因子维度控制模型表达能力maxIter是迭代次数通常 10 到 20 就收敛regParam是正则化系数防止过拟合alpha只在隐式反馈里用显式评分场景保持默认 1.0 即可。调参顺序我一般这样排先固定maxIter10、regParam0.1在rank的候选值 [10, 20, 50, 100] 里找 RMSE 最低的然后固定最优rank在regParam的 [0.01, 0.05, 0.1, 0.5] 里细调最后把maxIter加到 20 看是否还有提升。豆瓣评分数据上rank50、regParam0.1、maxIter15是一个比较稳的起点。from pyspark.ml.recommendation import ALS from pyspark.ml.evaluation import RegressionEvaluator train, test indexed.randomSplit([0.8, 0.2], seed42) als ALS( userColuser_idx, itemColmovie_idx, ratingColrating, rank50, maxIter15, regParam0.1, alpha1.0, coldStartStrategydrop, # 预测时丢弃冷启动样本 nonnegativeTrue, # 评分非负加约束更稳 seed42 ) model als.fit(train) predictions model.transform(test) evaluator RegressionEvaluator( metricNamermse, labelColrating, predictionColprediction ) rmse evaluator.evaluate(predictions) print(fTest RMSE {rmse:.4f})coldStartStrategydrop必须加。不加的话测试集里那些训练时没出现过的用户或电影会得到 NaN 预测值RMSE 直接变成 NaN。nonnegativeTrue对显式评分也有好处因为豆瓣评分是 1 到 5负的预测值没有物理意义加非负约束能轻微提升精度。3.2 模型持久化与跨版本加载的坑训练好的 ALS 模型用model.save()存到 HDFS 或本地加载时用ALSModel.load()。这里有一个版本兼容性问题Spark 2.x 和 3.x 的 ALS 模型格式不兼容用 2.4 训练的模型在 3.0 上加载会报NoSuchMethodError。如果团队里有人用不同版本的 Spark统一版本是唯一的后悔药。# 保存模型 model.save(hdfs:///models/als_douban) # 加载模型 from pyspark.ml.recommendation import ALSModel loaded_model ALSModel.load(hdfs:///models/als_douban) # 给指定用户批量生成推荐 user_recs loaded_model.recommendForAllUsers(10) user_recs.show(5, truncateFalse)recommendForAllUsers(10)返回每个用户 Top 10 推荐结果是一个数组列每个元素是 (movie_idx, rating) 结构。这个操作在用户量大时很吃内存因为要同时持有所有用户的推荐结果。我一般分片做先取出用户 ID 列表按每批 1000 个用户调用recommendForUserSubset避免 driver 端 OOM。另一个坑是recommendForAllUsers的默认并行度。它内部会做笛卡尔积式的候选生成如果spark.sql.shuffle.partitions设得太小任务会卡在最后一个 reduce 上。我习惯在调用前把 shuffle 分区数临时调大比如spark.conf.set(spark.sql.shuffle.partitions, 1000)跑完再调回来。4. 推荐结果评估与线上服务对接4.1 离线指标RMSE 之外的排序指标RMSE 衡量的是评分预测准不准但推荐系统真正关心的是排序对不对。一个 RMSE 很低的模型可能把用户喜欢的电影排在第 50 位那推荐效果依然很差。所以离线评估要加排序指标最常用的是 Hit Rate 和 NDCG。Hit Rate 的定义是在 Top-N 推荐里有多少用户命中了测试集中他们实际评过分的电影。计算时要注意测试集里的电影是用户已经看过的而推荐系统要推的是用户没看过的。严格来说离线评估应该留出用户最后一条评分做测试前面的做训练这样测试集里的电影对模型来说是“未来行为”。但豆瓣数据往往没有严格的时间切分我一般用随机切分然后在计算 Hit Rate 时排除训练集里已经出现过的 (user, movie) 对。# 计算 Hit Rate 的简化逻辑 # 先拿到每个用户的 Top 50 推荐 recs loaded_model.recommendForAllUsers(50) # 展开推荐结果得到 (user_idx, movie_idx) 对 from pyspark.sql.functions import explode rec_pairs recs.select(user_idx, explode(recommendations).alias(rec)) \ .select(user_idx, rec.movie_idx) # 测试集里的真实交互 test_pairs test.select(user_idx, movie_idx) # 命中推荐对和测试对重合 hits rec_pairs.join(test_pairs, [user_idx, movie_idx], inner) hit_rate hits.select(user_idx).distinct().count() / test.select(user_idx).distinct().count() print(fHit Rate50 {hit_rate:.4f})这个计算方式有个偏差测试集里的电影用户已经评过分而推荐列表里可能包含这些电影导致 Hit Rate 虚高。更严谨的做法是在生成推荐时过滤掉训练集里出现过的电影Spark ML 没有内置这个过滤需要自己写 UDF 或者用left_antijoin 排除。4.2 用 Flask 或 FastAPI 暴露推荐接口离线模型训练完线上服务一般用 Flask 或 FastAPI 包一层。核心逻辑是接收 user_id查索引表拿到 user_idx调用recommendForUserSubset再把 movie_idx 映射回 movie_id返回电影列表。索引表就是StringIndexer的labels数组存成 JSON 或数据库都行。from fastapi import FastAPI from pyspark.ml.recommendation import ALSModel from pyspark.sql import SparkSession app FastAPI() spark SparkSession.builder.appName(RecService).getOrCreate() model ALSModel.load(hdfs:///models/als_douban) # 假设索引映射已经加载到内存 user_labels [...] # StringIndexer 的 labels movie_labels [...] app.get(/recommend/{user_id}) def recommend(user_id: str, top_n: int 10): if user_id not in user_labels: # 冷启动返回热门电影 return {user_id: user_id, movies: popular_movies[:top_n]} user_idx user_labels.index(user_id) user_df spark.createDataFrame([(user_idx,)], [user_idx]) recs model.recommendForUserSubset(user_df, top_n).collect() if not recs: return {user_id: user_id, movies: popular_movies[:top_n]} movie_indices [r.movie_idx for r in recs[0].recommendations] movie_ids [movie_labels[i] for i in movie_indices] return {user_id: user_id, movies: movie_ids}这个服务有两个性能瓶颈一是每次请求都createDataFrameSpark 的 DataFrame 创建有固定开销二是recommendForUserSubset会触发一次 Spark 作业延迟在秒级。生产环境一般会把推荐结果预计算好存到 Redis 或 HBase接口只做查询。预计算的频率看业务需求豆瓣这种场景一天跑一次全量推荐就够了。注意FastAPI 里直接用 SparkSession 会有线程安全问题。SparkSession 是线程安全的但recommendForUserSubset并发调用时可能互相干扰。稳妥做法是加一个请求队列或者用ThreadPoolExecutor限制并发数。5. 避坑与排查ALS 训练和部署中的五个高频问题5.1 现象训练时报java.lang.OutOfMemoryError: Java heap space原因通常是rank设得太大或者spark.sql.shuffle.partitions太小导致单个分区数据量过大。ALS 在求解最小二乘时每个分区的用户因子矩阵大小是rank × rankrank200 时单个矩阵就占 200×200×8 字节几万个用户同时驻留内存很容易撑爆 executor。解决分三步先把rank降到 50 以下再把spark.sql.shuffle.partitions调到数据量的 2 到 3 倍最后给 executor 加内存spark.executor.memory8g起步。如果还不行检查是否有数据倾斜——某些热门电影被评了几十万次对应的 item 分区会特别大。可以用salting技术给热门 item 加随机前缀打散。5.2 现象RMSE 正常但推荐结果全是同一批电影这是典型的流行度偏差。ALS 在隐式反馈里会倾向于推荐热门物品显式评分场景下如果正则化不够也会出现类似问题。原因是热门电影的评分数据多因子向量被训练得更充分预测分数普遍偏高。解决办法有两个一是提高regParam比如从 0.1 提到 0.5压制热门物品的因子范数二是在推荐后做重排对每个用户已推荐的电影按热度降权公式是final_score pred_score / (rating_count ^ 0.5)。我一般两个一起用先调正则化再加重排。5.3 现象recommendForAllUsers跑几个小时不结束这个方法的内部实现是先给所有用户生成候选物品再做笛卡尔积式的评分预测。用户数 10 万、电影数 1 万时中间结果有 10 亿行即使 Spark 也扛不住。正确做法是分片把用户按user_idx % 100分成 100 片每片单独调recommendForUserSubset跑完合并。# 分片推荐避免单次作业过大 all_recs [] for i in range(100): user_subset indexed.filter(fuser_idx % 100 {i}) \ .select(user_idx).distinct() recs model.recommendForUserSubset(user_subset, 10) all_recs.append(recs) final_recs all_recs[0] for r in all_recs[1:]: final_recs final_recs.union(r)分片数按用户量调一般保证每片 1000 到 5000 个用户。分片太多会导致小文件问题分片太少又回到 OOM。5.4 现象线上服务返回的推荐和离线评估结果不一致最常见的原因是索引映射不一致。离线训练时StringIndexer的labels顺序和线上加载的不是同一份。比如离线用全量数据 fit线上用增量数据重新 fit同一部电影的 idx 就变了。解决方法是把StringIndexer的labels和模型一起保存线上加载时直接用保存的 labels不要重新 fit。Spark ML 的StringIndexerModel可以save和load但很多人只存了 ALS 模型忘了存索引器。5.5 现象冷启动用户请求延迟特别高冷启动用户不在索引表里代码会走热门电影兜底逻辑。但如果热门电影列表是从 HDFS 实时读的每次请求都触发一次文件读取延迟自然高。正确做法是把热门列表加载到内存用定时任务每天更新一次。# 启动时加载热门列表到内存 popular_movies spark.read.parquet(hdfs:///data/popular_movies) \ .collect() popular_movies [row.movie_id for row in popular_movies] # 定时刷新 import threading def refresh_popular(): global popular_movies popular_movies spark.read.parquet(hdfs:///data/popular_movies) \ .collect() popular_movies [row.movie_id for row in popular_movies] threading.Timer(86400, refresh_popular).start() refresh_popular()6. 进阶技巧用隐式反馈和混合推荐提升效果显式评分有个天然缺陷用户只对自己看过的电影评分没看过的电影不代表不喜欢。豆瓣上大量用户只看不评这些行为数据如果浪费掉很可惜。把“浏览过但没评分”当作隐式反馈用implicitPrefsTrue训练 ALS往往能覆盖更多用户。代价是评分值要重新定义评分 4 到 5 映射为 1.0评分 1 到 2 映射为 0.0没评分的默认也是 0.0。这样模型学的是“用户会不会喜欢”而不是“用户打几分”。# 构造隐式反馈数据 implicit_data indexed.withColumn( implicit_rating, when(col(rating) 4, 1.0).otherwise(0.0) ) als_implicit ALS( userColuser_idx, itemColmovie_idx, ratingColimplicit_rating, rank50, maxIter15, regParam0.1, implicitPrefsTrue, alpha40.0, # 隐式反馈的置信度参数通常设 10 到 40 coldStartStrategydrop, seed42 ) implicit_model als_implicit.fit(implicit_data)alpha在隐式反馈里控制置信度交互越多置信度越高。豆瓣数据上 alpha40 是一个经验值太小会导致模型忽略长尾行为太大又会让热门物品主导。可以按alpha 10 log(1 rating_count)动态设但 Spark ML 不支持逐样本 alpha只能全局设一个值。混合推荐是另一个提升方向ALS 负责协同过滤再单独训一个基于内容的模型比如用电影标签做 TF-IDF最后加权融合。权重用验证集调一般协同过滤占 0.7内容模型占 0.3。这个方案实现成本高但冷启动场景下内容模型能补上协同过滤的短板。我自己的习惯是先用显式 ALS 跑通全流程把 RMSE 和 Hit Rate 记下来作为基线然后切隐式反馈看 Hit Rate 有没有提升最后如果时间允许再加内容模型做融合。每一步都保留模型和评估结果方便回滚。这个方向值得做因为推荐系统的工程链路和调参经验可以迁移到电商、内容分发等场景Spark ML 的 ALS 只是入口背后的特征工程、离线评估、线上服务才是真正吃功夫的地方。希望帮到你。本文还有配套的精品资源点击获取
