基于Spark的电影推荐系统实战：从爬虫采集到ALS模型与Web部署

简介这是一套基于Spark的电影推荐系统完整工程整合爬虫采集、Web展示、后台管理与推荐算法模块适合计算机相关专业学生用于毕业设计、课程设计或项目实战。压缩包共1417个文件约59.61MB包含232个HTML、226个CSS、218个JS构成的前端页面108个Java与51个Scala编写的后端及Spark推荐逻辑另有Python爬虫脚本、SQL初始化数据、配置文件与文档等结构覆盖数据采集、存储、模型训练到Web交互的完整链路。项目已通过导师指导与答辩评审代码经过运行测试功能稳定已有59人浏览学习。压缩包内含可运行的完整项目、详细说明文档还附带PPT汇报、相关论文等辅助材料便于参考实现和二次开发也适合从零理解推荐系统落地流程。1. 拿到“基于Spark的电影推荐系统”压缩包先想清楚要跑通什么很多人在找实战项目时看到“基于 Spark 的电影推荐系统”这个组合第一反应是堆料爬虫、web 项目、后台管理系统、推荐算法全塞在一个压缩包里。但如果你按顺序把每一条链路跑通得到的是一条完整的数据产品流水线——爬虫抓取电影和评分数据入库Spark 做 ETL 清洗ALS 协同过滤训练推荐模型Web 网站对外展示结果后台管理系统做数据维护和监控。这套方案值得投入是因为它覆盖了招聘里最常被问到的几项技能Python 爬虫、Spark 数据处理、推荐算法、Web 开发。适合三类人准备毕业设计的学生、想给简历加一个闭环项目的新人、刚学完 Spark 缺实战目标的人。下面按数据流向拆开讲。2. 爬虫采集与数据入库用 SQLAlchemy 把数据先落稳2.1 先定义表结构电影、评分、行为日志爬虫是整个系统的起点但很多人拿到项目第一件事是去跑 Spark结果发现库里没数据回头再补爬虫。正确顺序是先定表结构再写爬虫。推荐系统最少需要两张表电影表movies和评分表ratings。电影表存电影基础信息评分表存用户对电影的反馈这是 ALS 算法唯一需要的输入。from sqlalchemy import create_engine, Column, Integer, Float, String from sqlalchemy.ext.declarative import declarative_base Base declarative_base() class Movie(Base): __tablename__ movies movie_id Column(Integer, primary_keyTrue) title Column(String(255), nullableFalse) genres Column(String(255)) # 类型字段多个类型用 | 分隔 year Column(Integer) class Rating(Base): __tablename__ ratings id Column(Integer, primary_keyTrue, autoincrementTrue) user_id Column(Integer, indexTrue) movie_id Column(Integer, indexTrue) rating Column(Float) # 评分范围 0.5 ~ 5.0 ts Column(Integer) # unix 秒级时间戳为什么用indexTrue后续 Spark ETL 和 SQL 查询会频繁按 user_id、movie_id 过滤和 join索引能明显加快读取。为什么ts用 Integer 而不是 DateTimeSpark 处理 unix 时间戳更顺手一行from_unixtime就能转换省去跨库时区问题。这里有一个常见翻车点如果连接 MySQL连接串要带charsetutf8mb4否则电影标题里的特殊字符会直接入库失败。2.2 requests BeautifulSoup 抓取session 复用是关键爬虫抓取网站数据时最容易被忽视的不是解析规则而是请求会话的复用。每次请求都新建连接抓几千页后 TCP 连接开销会拖慢整个任务。常见做法是用一个requests.Session()承载所有请求配合浏览器风格的请求头。import requests from bs4 import BeautifulSoup from sqlalchemy.orm import sessionmaker HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Referer: https://example.com/, } def fetch_page(offset): url fhttps://example.com/movies?offset{offset} resp requests.get(url, headersHEADERS, timeout10) resp.raise_for_status() soup BeautifulSoup(resp.text, html.parser) rows [] for node in soup.select(.movie-card): rows.append({ movie_id: int(node[data-movie-id]), title: node.select_one(.title).text.strip(), genres: node.select_one(.genres).text.strip(), year: int(node.select_one(.year).text.strip()), }) return rows这段代码的工程点有三个User-Agent要伪装成浏览器分页参数用offset而不是page很多列表页的分页模型是偏移量CSS 选择器要跟随目标站点的实际结构调整。调参重点timeout10比 requests 默认值更能容忍慢站抓取时把连接放进 Session 而不是每次 get 都新建。数据抓下来后要落库SQLAlchemy 的merge是增量更新的地基engine create_engine( mysqlpymysql://rec:rec123localhost/recommend?charsetutf8mb4, pool_size10, pool_recycle3600 ) Session sessionmaker(bindengine) def save_movies(movies): with Session() as s: for m in movies: s.merge(Movie(**m)) # 按主键存在则更新不存在则插入 s.commit()pool_size10提高连接池上限避免并发写库时排队pool_recycle3600让连接每小时重建一次防止 MySQL 的wait_timeout把空闲连接断开。初学者最常见的做法是每次全量 delete 再 insert数据量小看不出来等爬到几万条再重跑一次你就知道什么叫后悔药没得吃。2.3 增量更新与断点续爬把进度落盘爬虫跑了几小时后崩溃、网络断开、目标站点临时返回 500这些都是常态。给爬虫装“后悔药”的核心是进度必须落盘不能只存在内存里。用一个小 JSON 文件记录当前抓到的位置脚本重启后从断点继续。import json, time STATE_FILE crawl_state.json def load_state(): try: with open(STATE_FILE) as f: return json.load(f) except FileNotFoundError: return {max_movie_id: 0, page: 0} def crawl_incremental(batch50, sleep_secs2): state load_state() offset state[page] while True: rows fetch_page(offset) if not rows: break unseen [r for r in rows if r[movie_id] state[max_movie_id]] if unseen: save_movies(unseen) state[max_movie_id] max(r[movie_id] for r in unseen) state[page] offset with open(STATE_FILE, w) as f: json.dump(state, f) offset batch time.sleep(sleep_secs)逻辑说明每一批抓完先把page和max_movie_id写回文件再继续下一页。这样无论脚本怎么退出重跑时都能从上次的位置继续。movie_id max_movie_id是天然的去重条件比先查数据库再决定插入要快得多。sleep_secs是限速参数一般设 1 到 3 秒比请求头伪装更管用的反反爬手段就是降低请求频率。爬虫技术抓取网站数据时频率才是最容易被风控盯上的特征。3. Spark 集群搭建与 ETL 清洗把原始数据洗成评分表3.1 spark 集群搭建的最小成本路径先 local 后 standalone很多教程一上来就让你准备三台虚拟机搭 HA 集群对这个项目来说完全没必要。Spark 的应用入口是SparkSession本地local[*]模式就能跑通 ETL 和 ALS 训练等代码稳定了再上 standalone 集群验证资源分配。spark 集群搭建的正确顺序是先在单机把逻辑跑通再考虑分布式。# 解压 Spark 后进入 conf 目录 cp spark-env.sh.template spark-env.sh cat spark-env.sh EOF export JAVA_HOME/usr/lib/jvm/java-8-openjdk-amd64 export SPARK_MASTER_HOST192.168.1.10 export SPARK_WORKER_CORES2 export SPARK_WORKER_MEMORY4g EOF cp spark-defaults.conf.template spark-defaults.conf cat spark-defaults.conf EOF spark.master spark://192.168.1.10:7077 spark.serializer org.apache.spark.serializer.KryoSerializer EOF启动命令是$SPARK_HOME/sbin/start-master.sh和start-worker.sh spark://192.168.1.10:7077用jps能看到 Master 和 Worker 进程即成功。这里的两个关键选型JAVA_HOME指向 JDK 8 或 11Spark 3.x 配 JDK 17 会出现一堆莫名的反射异常KryoSerializer减少 shuffle 阶段的序列化开销ALS 这类需要大量 shuffle 的任务收益明显。3.2 用 Spark SQL 做 ETL去重、过滤、类型收敛从数据库或 CSV 读出来的原始数据是不能直接训练的。评分可能有空值、超范围值、重复记录电影标题可能有编码问题。ETL 阶段把这些脏数据清掉。from pyspark.sql import SparkSession from pyspark.sql.functions import col, when spark SparkSession.builder \ .appName(movie_etl) \ .master(local[*]) \ .config(spark.sql.shuffle.partitions, 4) \ .getOrCreate() ratings spark.read \ .option(header, True) \ .option(inferSchema, True) \ .csv(data/ratings.csv) ratings ratings \ .filter(col(user_id).isNotNull()) \ .filter(col(movie_id).isNotNull()) \ .filter(col(rating).between(0.5, 5.0)) \ .dropDuplicates([user_id, movie_id, ts])inferSchemaTrue让 Spark 自动推断列类型省去手写 schema 的麻烦。between(0.5, 5.0)是域约束比单纯的非空过滤更严格能把爬虫抓到的异常评分挡在训练集之外。dropDuplicates按用户、电影、时间戳三列去重防止爬虫重复抓取造成同一用户对同一电影有多条评分。spark.sql.shuffle.partitions4是给本地调试用的默认 200 个分区在小数据集上会产生大量空 task 和小文件。3.3 训练前最后一次加工把字符串 ID 转成数值索引ALS 要求 user 和 item 列是数值型但爬虫抓下来的 user_id 可能是字符串。这时候用StringIndexer做编码。另外评分表里可能混着没有电影基础信息的脏数据要 join 掉。from pyspark.ml import Pipeline from pyspark.ml.feature import StringIndexer user_indexer StringIndexer(inputColuser_id, outputColuser_idx) \ .setHandleInvalid(skip) movie_indexer StringIndexer(inputColmovie_id, outputColmovie_idx) \ .setHandleInvalid(skip) pipeline Pipeline(stages[user_indexer, movie_indexer]) train pipeline.fit(ratings).transform(ratings) \ .select(user_idx, movie_idx, rating) movies spark.read.csv(data/movies.csv, headerTrue) train train.join(movies.select(movie_id), movie_id, inner)setHandleInvalid(skip)很关键当新数据里出现训练时没见过的 IDSpark 默认会抛错skip 会让这些记录直接跳过不影响整体流程。最后的 join 意图是去掉“有评分但没抓到电影资料”的记录这类数据对 ALS 来说是纯噪声。到这一步训练表就是干净的三列user_idx、movie_idx、rating。4. 推荐算法与 Web 产品化ALS 训练、冷启动与后台管理系统4.1 ALS 协同过滤三个必调参数ALS交替最小二乘法是 Spark 自带矩阵分解实现适合显式评分数据。它把用户和物品映射到同一个隐因子空间用低秩矩阵逼近评分矩阵。对初学者来说不需要深挖数学推导但三个参数必须理解。rank是隐因子维度控制模型的表达能力一般取 10 到 50regParam是正则化系数防止过拟合常用 0.01 到 0.1maxIter是迭代次数10 到 20 轮足够收敛。from pyspark.ml.recommendation import ALS als ALS( userColuser_idx, itemColmovie_idx, ratingColrating, rank20, maxIter15, regParam0.05, implicitPrefsFalse, coldStartStrategydrop ) model als.fit(train)implicitPrefsFalse表示这是显式评分如果你的数据源是点击、浏览这类隐式反馈要设成 True并且额外调alpha参数。coldStartStrategydrop让预测阶段遇到未知用户或电影时直接丢弃而不是返回 NaN——这个参数不设在线推理时你会被 NaN 结果折腾到怀疑人生。先手动跑几组 rank 和 regParam看 RMSE 的变化趋势稳定后再用ParamGridBuilder做网格搜索。4.2 训练与评估RMSE 不是唯一的验收标准划分训练集和测试集用 RMSE 做离线评估from pyspark.ml.evaluation import RegressionEvaluator (train_set, test_set) train.randomSplit([0.8, 0.2], seed42) evaluator RegressionEvaluator( metricNamermse, labelColrating, predictionColprediction ) rmse evaluator.evaluate(model.transform(test_set)) print(fRMSE {rmse:.4f})randomSplit按 8:2 切分seed42保证可复现。但 RMSE 低不代表推荐效果好——模型完全可以把所有预测值压到全局平均附近RMSE 看着还行推荐结果却毫无个性。所以离线评估之外必须做两件事一是人工抽检随机挑几个用户的 Top-N 推荐看是不是真的符合口味二是算覆盖率推荐结果里出现过多少不同的电影如果永远只有头部热门片说明模型已经退化了。4.3 Web 网站接入推荐结果模型文件怎么被调用训练好的模型要暴露给 Web 项目常见做法是用 Flask 写一个推荐接口加载 Spark 模型文件接收用户 ID 返回 Top-N 电影。from flask import Flask, jsonify from pyspark.sql import SparkSession from pyspark.ml.recommendation import ALSModel spark SparkSession.builder \ .appName(rec_api) \ .master(local[2]) \ .getOrCreate() model ALSModel.load(models/als_model_v3) app Flask(__name__) app.route(/api/v1/recommend/int:user_idx) def recommend(user_idx): df spark.createDataFrame([(user_idx,)], [user_idx]) recs model.recommendForUserSubset(df, 20) if recs.count() 0: return jsonify({items: [], strategy: cold_start}) items recs.collect()[0][recommendations] movie_ids [r.movie_idx for r in items] return jsonify({items: movie_ids, strategy: als})每次请求都createDataFrame再collect其实对中小流量完全够用。如果 QPS 高把 Top-N 结果在训练时预生成写进 RedisWeb 层直接读缓存。这里要养成一个习惯模型文件带版本号不要覆盖models/als_model这个路径——Web 服务热加载时可能读到写了一半的文件线上推荐就会短暂失效。模型文件路径和版本号是推荐系统里的一个常见黑匣子出了问题先看这里。4.4 后台管理系统看板、日志和模型版本这个项目里的后台管理系统不需要做成一个重型的 Vue3 中后台。它的核心职责是数据维护和运行监控。用 Flask-Admin 或 Django Admin 十分钟就能生成 CRUD 页面。后台模块作用对应数据表用户管理查看用户列表、禁用异常账号users电影管理维护电影基础信息和上架状态movies推荐日志记录每次推荐结果与后续点击rec_logs爬虫监控查看抓取进度、错误次数crawl_state最应该花时间做的不是花哨的界面而是推荐日志。记录“给哪个用户推了哪些电影、用户有没有点击、点击后有没有评分”这些日志是下一轮训练的数据资产。没有日志反馈的推荐系统永远只能离线调参无法闭环。如果你的简历要写 Vue3 后台管理系统那就单独让后端出 REST API前端用 Vue3 接但别让后台的开发量吃掉推荐系统本身的时间。5. 避坑跑这个 Spark 电影推荐系统必踩的 5 个坑5.1 Spark、Scala、JDK 版本不匹配黑匣子一样的 NoSuchMethodError现象spark-submit或 Web 服务启动时报java.lang.NoSuchMethodError或者 pyspark 一运行就崩堆栈里全是 Scala 包名。原因Spark 3.x 用 Scala 2.12 编译而项目里某个第三方 Jar 是 Scala 2.11 编的或者 JDK 版本过高反射调用失败。解决先跑spark-submit --version确认 Spark 自带的 Scala 版本用 pip 安装与 Spark 二进制相同版本的 pysparkJDK 固定在 8 或 11不要追新。这个项目里的 Scala 版本匹配问题是环境类报错里出现频率最高的。5.2 爬虫返回 403抓着抓着全是空数据现象爬虫跑了几百页后raise_for_status()抛 403或者返回的列表突然为空。原因请求频率太高目标站的风控把 IP 或会话标记了User-Agent 太像脚本缺少 Referer 等关键头。解决sleep_secs调到 2 秒以上准备三五个 User-Agent 随机切换加指数退避重试。爬虫技术抓取网站数据时最重要的不是解析代码而是限速和重试策略。记住一个原则增量爬永远比全量重爬安全——全量重爬会把之前积累的风控风险重新触发一遍。5.3 SQLAlchemy 连接池耗尽爬虫最常见的翻车点现象爬虫跑到一半报sqlalchemy.exc.TimeoutError: QueuePool limit of size 5 overflow 10 reached。原因代码里每个循环都调用create_engine()新建连接连接没释放池被占满。解决整个项目只创建一个全局engine用with Session() as s确保 session 自动关闭配合pool_size10, pool_recycle3600。还有一个隐蔽点爬虫进程如果用了多线程每个线程都要注意 session 不要跨线程传递否则连接归还时机不确定。5.4 Windows 本地跑 Spark 缺 winutils现象Spark 在 Windows 上一启动就报Failed to locate the winutils binary in the Hadoop binaries。原因Spark 依赖 Hadoop native 库Windows 下没有这个二进制文件。解决下载与 Hadoop 版本对应的winutils.exe放到%HADOOP_HOME%\bin目录并配置环境变量或者更省心的做法是直接用 Docker 跑 Spark 镜像。血泪经验别在 Windows 上和 Spark 死磕本地开发用 Docker线上用 Linux能省出大半天时间。5.5 推荐结果一直不变模型根本没重训现象新用户、新评分都进库了Web 端返回的推荐列表和一个月前一模一样。原因训练脚本只在第一次手动执行过Web 服务加载的始终是旧模型文件或者评分数据变了但 ETL 没把新数据喂进训练集。解决把训练做成定时任务每天凌晨执行“拉新数据 → ETL → 训练 → 模型版本号 1 → 通知 Web 加载新模型”。模型目录按als_model_v3、als_model_v4命名Web 端启动时读取最新版本或者做个简单的文件指纹校验发现变化再 reload。6. 验证与进阶把推荐系统从“能跑”变“可信”6.1 离线指标与人工抽检RMSE 只能反映全局误差推荐质量还要看 Top-N 命中率。一个简单可执行的验证方法从测试集里取出每个用户的真实高分电影看推荐的前 10 个里命中了几个计算 Precision10。如果命中率长期低于 5%说明模型基本没学到用户偏好问题多半出在数据质量而不是算法。hit 0 total 0 for user_idx in test_users: top_n recommend(user_idx, 10) real get_user_high_rated(user_idx, threshold4.0) hit len(set(top_n) set(real)) total 10 precision hit / total print(fPrecision10 {precision:.3f})6.2 定时重训与日志回流离线模型要变成可持续运行的系统需要一个定时任务0 3 * * * cd /opt/movie_rec ./retrain.sh logs/retrain.log 21retrain.sh里做四件事拉取新数据、跑 ETL、训练 ALS、模型版本号加一。推荐日志每天导出拼上用户行为数据下一轮训练就能用上真实的点击反馈。换模型前先离线对比新旧两个版本在测试集上的 Precision10确认不跌再切换给自己留个后悔药。我以前拿到这类项目包第一件事是打开 README第二件事就是配环境结果总在版本和连接池上浪费一晚上。现在的习惯是先跑通 ETL、再训练出模型最后才碰 Web 和后台——数据不动上面全是空中楼阁。把顺序反过来你能省下大量查错时间。希望帮到你。本文还有配套的精品资源点击获取