简介：这份资源是一套基于Spark的电影推荐系统完整项目包，面向计算机、人工智能、通信工程等专业的在校学生与教师，可用于毕业设计、课程设计、项目立项演示或自学进阶。项目整合了爬虫数据采集、Web网站、后台管理系统与Spark推荐算法四大模块，并附有详细文档，帮助读者理解从数据抓取到推荐落地的全流程。压缩包共1417个文件，约59.61MB，以html、css、js等前端页面文件，java、scala、py等后端与算法源码，以及png、gif、jpg等图片素材为主，另含xml、json、parquet、properties等配置与数据文件，目录结构清晰，便于按模块检索。目前已有60人学习下载。项目源码经测试运行成功，答辩评审达95分，读者可据此掌握推荐系统架构设计、前后端交互与Spark算法实现，并在此基础上修改扩展功能。
1. 从零搭一套电影推荐系统：为什么“全链路”比单点模型更值得做
很多人第一次接触推荐系统，都是从调一个协同过滤算法开始的：拿一份评分数据，跑一遍 ALS，算出 RMSE，觉得大功告成。但真到落地环节就会发现，模型只是整条链路里最不值钱的一环——没有数据来源，模型是空转的；没有前端承接，推荐结果没人看得到；没有后台管理，运营连一部电影都上不了架。这套“基于 Spark 的电影推荐系统”之所以值得认真做一遍，恰恰因为它把爬虫、Web 网站、后台管理系统和 Spark 推荐引擎串成了一条完整链路，是一个能写进简历、也能真正跑起来的工程闭环。
它适合三类人：想从“会调库”进阶到“能交付系统”的学生；需要一套可演示、可扩展架构的初中级工程师；以及想理解推荐系统数据流全貌的转行者。整条链路的核心逻辑是：爬虫负责把电影和评分数据抓回来，Spark 负责离线训练和生成推荐，Web 网站负责把推荐结果呈现给用户，后台管理系统负责数据维护和运营。四者缺一不可，而 Spark 是把它们粘合起来的中枢。下面我按真实搭建顺序，把每一步拆开讲清楚。
2. 数据从哪来：爬虫采集与清洗的完整落地路径
推荐系统的天花板由数据决定，所以第一步永远是解决“数据从哪来”。电影类数据通常包含两类：电影元信息（片名、类型、导演、上映年份、简介）和用户行为数据（评分、评论、观看记录）。元信息可以靠爬虫从公开的电影资料站点采集，行为数据则往往需要自己构造或从公开数据集迁移。这里我用一个通用的爬虫结构来演示，具体站点结构请按目标站点的实际 DOM 调整。
2.1 用 requests + BeautifulSoup 抓取电影元信息
先写一个最小可用的采集脚本，把列表页和详情页分开处理。列表页负责拿到详情页链接，详情页负责解析字段。这种“两级抓取”是爬虫里最稳的结构，避免一次性把逻辑写死在一个函数里。
import requests from bs4 import BeautifulSoup import time import csv import random HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36" } def fetch(url, retry=3): """带重试的请求封装，失败后指数退避""" for i in range(retry): try: resp = requests.get(url, headers=HEADERS, timeout=10) if resp.status_code == 200: resp.encoding = resp.apparent_encoding return resp.text except requests.RequestException as e: print(f"第{i+1}次请求失败: {e}") time.sleep(2 ** i) # 1s, 2s, 4s 退避 return None def parse_list(html): """从列表页提取详情页链接""" soup = BeautifulSoup(html, "html.parser") links = [] for a in soup.select("div.item a.title"): href = a.get("href") if href: links.append(href) return links def parse_detail(html): """从详情页解析电影字段，字段名按目标站点调整""" soup = BeautifulSoup(html, "html.parser") movie = {} movie["title"] = soup.select_one("h1 span").get_text(strip=True) if soup.select_one("h1 span") else "" movie["year"] = soup.select_one("span.year").get_text(strip=True) if soup.select_one("span.year") else "" movie["rating"] = soup.select_one("strong.rating_num").get_text(strip=True) if soup.select_one("strong.rating_num") else "" movie["genres"] = "|".join([g.get_text(strip=True) for g in soup.select("span.genre")]) movie["summary"] = soup.select_one("span[property='v:summary']").get_text(strip=True) if soup.select_one("span[property='v:summary']") else "" return movie if __name__ == "__main__": base = "https://example-movie-site.com/top250?start={}" all_movies = [] for page in range(0, 250, 25): html = fetch(base.format(page)) if not html: continue for link in parse_list(html): detail_html = fetch(link) if detail_html: all_movies.append(parse_detail(detail_html)) time.sleep(random.uniform(1, 3)) # 随机间隔，降低被封风险 print(f"已采集 {len(all_movies)} 条") with open("movies.csv", "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=["title", "year", "rating", "genres", "summary"]) writer.writeheader() writer.writerows(all_movies)这段代码的关键点有三个。第一，fetch里做了重试和指数退避，网络抖动是爬虫最常见的翻车点，没有重试机制跑一半就断。第二，time.sleep(random.uniform(1, 3))是随机间隔，固定间隔容易被识别为机器行为。第三，字段解析全部用select_one加空值判断，避免某个字段缺失导致整条记录报错。参数上，timeout=10是请求超时，太小会误判失败，太大则卡住流程；retry=3是重试次数，一般 3 次足够，再多说明目标站点已经不可用。
2.2 数据清洗：把脏数据挡在推荐引擎之外
爬回来的数据几乎不可能直接进模型。常见问题包括：评分字段带“分”字、年份带括号、类型字段分隔符不统一、简介里有换行和多余空格。清洗这一步做不干净，后面 Spark 训练出来的推荐结果就是玄学。
import pandas as pd import re def clean_movies(path): df = pd.read_csv(path) # 评分：去掉非数字字符，转 float df["rating"] = df["rating"].astype(str).str.extract(r"(\d+\.?\d*)").astype(float) # 年份：提取四位数字 df["year"] = df["year"].astype(str).str.extract(r"(\d{4})").astype("Int64") # 类型：统一分隔符为竖线，去空格 df["genres"] = df["genres"].astype(str).str.replace(r"[\s/、]+", "|", regex=True).str.strip("|") # 简介：去换行、去多余空格 df["summary"] = df["summary"].astype(str).str.replace(r"\s+", " ", regex=True).str.strip() # 丢弃关键字段缺失的行 df = df.dropna(subset=["title", "rating"]) df = df[df["rating"] > 0] return df if __name__ == "__main__": df = clean_movies("movies.csv") df.to_csv("movies_clean.csv", index=False, encoding="utf-8") print(f"清洗后剩余 {len(df)} 条，评分范围 {df['rating'].min()} - {df['rating'].max()}")清洗逻辑里，str.extract(r"(\d+\.?\d*)")用正则把评分里的数字抠出来，比直接replace更稳，因为不同站点的评分格式不一样。dropna只针对 title 和 rating，因为这两个字段是推荐系统的最小依赖，缺了就没法用。年份用Int64而不是int，是为了保留空值而不报错。清洗完一定要打印一下数量和评分范围，这是判断数据质量的第一个信号——如果评分范围是 0 到 0，说明正则没匹配上，得回去看原始数据长什么样。
3. Spark 推荐引擎：ALS 训练、调参与离线推荐的工程细节
数据准备好之后，就进入 Spark 的主场。电影推荐最常用的算法是 ALS（交替最小二乘），它属于协同过滤里的矩阵分解方法，核心思想是把用户-物品评分矩阵拆成两个低维矩阵，用隐向量表示用户和物品，再通过内积预测缺失评分。Spark MLlib 里的ALS实现成熟、分布式、支持隐式反馈，是这类系统的默认选择。
3.1 用 Spark 读取数据并构建评分矩阵
ALS 需要的数据格式是(userId, itemId, rating)三元组。电影元信息里的评分是“电影平均分”，不是“某个用户对某电影的评分”，所以真实系统里必须构造用户行为数据。常见做法有两种：一是用公开的评分数据集（如 MovieLens 格式）迁移，二是用规则模拟生成。这里演示从 CSV 读取并构建训练集的标准流程。
from pyspark.sql import SparkSession from pyspark.sql.functions import col, when, rand from pyspark.ml.recommendation import ALS from pyspark.ml.evaluation import RegressionEvaluator spark = SparkSession.builder \ .appName("MovieRecommender") \ .config("spark.sql.shuffle.partitions", "200") \ .config("spark.executor.memory", "4g") \ .getOrCreate() # 读取评分数据，格式：userId,movieId,rating,timestamp ratings = spark.read.csv("ratings.csv", header=True, inferSchema=True) ratings = ratings.select( col("userId").cast("int"), col("movieId").cast("int"), col("rating").cast("float") ).dropna() # 划分训练集和测试集 train, test = ratings.randomSplit([0.8, 0.2], seed=42) print(f"训练集 {train.count()} 条，测试集 {test.count()} 条")spark.sql.shuffle.partitions设成 200 是经验值，默认的 200 在中小数据集上够用，数据量再大要往上调，否则 shuffle 阶段会成为瓶颈。randomSplit的seed固定住，保证每次划分一致，方便复现。训练集和测试集的比例 8:2 是推荐系统里比较稳的划分，数据量特别小的时候可以调到 7:3。
3.2 ALS 参数怎么调：rank、regParam、maxIter 的取舍
ALS 有三个核心参数：rank（隐向量维度）、regParam（正则化系数）、maxIter（迭代次数）。这三个参数没有万能值，但有一套可复用的调参思路。
| 参数 | 含义 | 常用范围 | 调大后的影响 |
|---|---|---|---|
| rank | 隐向量维度 | 10 ~ 200 | 表达能力增强，但容易过拟合，训练变慢 |
| regParam | 正则化系数 | 0.01 ~ 1.0 | 抑制过拟合，太大导致欠拟合 |
| maxIter | 迭代次数 | 10 ~ 30 | 收敛更充分，但收益递减，耗时增加 |
| alpha | 隐式反馈置信度 | 1.0 ~ 40.0 | 仅隐式反馈时使用，越大越信任正样本 |
als = ALS( userCol="userId", itemCol="movieId", ratingCol="rating", rank=50, regParam=0.1, maxIter=15, nonnegative=True, coldStartStrategy="drop" ) model = als.fit(train) predictions = model.transform(test) evaluator = RegressionEvaluator( metricName="rmse", labelCol="rating", predictionCol="prediction" ) rmse = evaluator.evaluate(predictions) print(f"RMSE = {rmse:.4f}")nonnegative=True强制隐向量非负，在电影评分场景下通常能提升可解释性，因为评分本身是正向的。coldStartStrategy="drop"是关键，测试集里可能出现训练集没见过的用户或物品，ALS 对这类冷启动样本会预测出 NaN，不 drop 的话 RMSE 直接变成 NaN，这是新手最容易踩的坑之一。RMSE 在 0.8 到 1.0 之间通常算正常，低于 0.7 要警惕过拟合，高于 1.2 说明模型没学到东西。
3.3 生成推荐结果并落库
训练完模型后，要给每个用户生成 Top-N 推荐，并把结果写到数据库或文件，供 Web 端读取。
# 为每个用户生成 Top 10 推荐 user_recs = model.recommendForAllUsers(10) # 展开成 (userId, movieId, rating) 的扁平结构 from pyspark.sql.functions import explode flat_recs = user_recs.select( col("userId"), explode(col("recommendations")).alias("rec") ).select( col("userId"), col("rec.movieId").alias("movieId"), col("rec.rating").alias("score") ) # 写入 MySQL，供 Web 端查询 flat_recs.write \ .format("jdbc") \ .option("url", "jdbc:mysql://localhost:3306/movie_rec") \ .option("dbtable", "user_recommendations") \ .option("user", "root") \ .option("password", "your_password") \ .option("batchsize", "1000") \ .mode("overwrite") \ .save() print("推荐结果已写入数据库")recommendForAllUsers(10)返回的是一个数组列，必须用explode展开成行，否则没法直接写关系型数据库。batchsize=1000是 JDBC 写入的批量大小，太小会导致频繁网络往返，太大容易内存溢出，1000 是折中值。mode("overwrite")每次全量覆盖，适合离线批处理场景；如果要做增量更新，改成append并配合时间分区。
4. Web 网站与后台管理：让推荐结果真正被看见
模型跑通只是完成了“后台”，用户看不到推荐结果，这套系统就只是个脚本。Web 网站负责展示，后台管理系统负责维护，两者共用同一套数据库，是这套架构里最容易被低估的部分。
4.1 Web 端推荐展示的最小实现
Web 端通常用 Flask 或 Django 起一个轻量服务，从数据库读取推荐结果，渲染到页面上。核心逻辑是：用户登录后拿到 userId，查user_recommendations表，关联movies表拿到电影详情，返回给前端。
from flask import Flask, render_template, session, redirect import pymysql app = Flask(__name__) app.secret_key = "your_secret_key" def get_db(): return pymysql.connect( host="localhost", user="root", password="your_password", database="movie_rec", charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor ) @app.route("/recommend") def recommend(): if "user_id" not in session: return redirect("/login") uid = session["user_id"] conn = get_db() with conn.cursor() as cur: cur.execute(""" SELECT m.title, m.year, m.genres, m.summary, r.score FROM user_recommendations r JOIN movies m ON r.movieId = m.movieId WHERE r.userId = %s ORDER BY r.score DESC LIMIT 20 """, (uid,)) movies = cur.fetchall() conn.close() return render_template("recommend.html", movies=movies)这里用JOIN把推荐结果和电影详情关联起来，避免在 Python 里做二次查询。ORDER BY r.score DESC保证推荐按分数排序，LIMIT 20控制单页数据量。注意session里存的是 userId，不是用户名，因为推荐表用的是 userId 关联，用用户名会多一次查询。
4.2 后台管理系统的功能边界
后台管理系统不需要花哨，但必须覆盖四件事：电影数据的增删改查、用户管理、推荐结果查看、爬虫任务触发。用同一套 Flask 加一个管理蓝图就能实现，关键是把权限控制住，普通用户不能进后台。
| 功能模块 | 核心操作 | 数据表 |
|---|---|---|
| 电影管理 | 新增/编辑/下架电影 | movies |
| 用户管理 | 查看用户、禁用账号 | users |
| 推荐查看 | 按用户查推荐列表 | user_recommendations |
| 任务触发 | 手动触发爬虫和训练 | task_log |
后台的“任务触发”是最容易被忽略但最实用的功能。离线推荐不是实时的，运营需要能手动跑一次爬虫或重新训练模型，而不是每次都登服务器敲命令。用一个简单的任务表记录触发时间和状态，配合后台按钮调用脚本即可。
5. 避坑与排查：这套系统最容易翻车的五个地方
5.1 爬虫跑一半被封，数据缺了一大块
现象：采集到几百条后请求全部返回 403 或空页面。原因：请求频率过高或 User-Agent 单一，被目标站点识别。解决：加随机间隔、轮换 User-Agent、必要时用代理池，并且把已采集的数据落盘，支持断点续采，不要每次都从头跑。
5.2 ALS 预测出 NaN，RMSE 直接报错
现象：evaluator.evaluate返回 NaN。原因：测试集里有训练集没出现过的 userId 或 movieId，ALS 对冷启动样本预测为 NaN。解决：设置coldStartStrategy="drop"，或者在划分数据集时保证用户和物品在训练集中都出现过。
5.3 推荐结果全是同一部电影
现象：所有用户的 Top 10 推荐高度重合。原因：数据稀疏，热门电影主导了隐向量，模型退化成“推荐最热”。解决：提高regParam抑制热门偏置，或者在评分数据里对热门电影做降权，也可以引入基于内容的特征做混合推荐。
5.4 Spark 任务内存溢出，Executor 频繁挂掉
现象：训练阶段报OutOfMemoryError，Executor 被 kill。原因：rank设得太大、数据没分区、或者spark.executor.memory不够。解决：先把rank降到 50 以下，检查spark.sql.shuffle.partitions是否合理，必要时增加 Executor 内存并开启spark.memory.fraction调优。
5.5 Web 端查询推荐结果慢到超时
现象：推荐页加载超过 5 秒。原因：user_recommendations表没建索引，或者每次查询都全表扫描。解决：在userId和movieId上建联合索引，推荐结果表按 userId 做分区，查询时只取需要的列，不要SELECT *。
6. 进阶技巧：把离线推荐做成可迭代的工程习惯
这套系统跑通之后，真正拉开差距的不是模型多复杂，而是迭代效率。我自己的习惯是每次调整 ALS 参数前，先把当前 RMSE、训练耗时、推荐覆盖率记到一张实验表里，跑完新参数再对比。没有这张表，调参就是凭感觉，跑十次也说不清哪次更好。
import json, time def log_experiment(params, rmse, duration, path="experiments.jsonl"): record = { "ts": time.strftime("%Y-%m-%d %H:%M:%S"), "params": params, "rmse": round(rmse, 4), "duration_sec": round(duration, 1) } with open(path, "a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") # 用法：每次训练后调用 log_experiment({"rank": 50, "regParam": 0.1, "maxIter": 15}, rmse, 128.5)另一个实用技巧是把推荐结果做分层：Top 10 走 ALS，Top 10 到 Top 30 用基于类型的相似度补位。ALS 在数据稀疏时对长尾物品的预测不稳定，用类型相似度兜底能明显提升推荐列表的多样性。具体做法是拿用户历史高分电影的类型分布，去匹配同类型里评分靠前但 ALS 没推的电影，合并去重后按分数排序。
最后一个习惯：每次重新训练模型前，先备份当前的user_recommendations表。离线推荐是全量覆盖的，一旦新模型效果变差，没有备份就只能等下一次训练，这个后悔药我吃过不止一次。把备份做成脚本里的一行CREATE TABLE ... AS SELECT，成本极低，但能救命。
这套系统真正的价值不在于用了多新的算法，而在于它把数据采集、模型训练、服务展示和运营管理串成了一条能持续迭代的链路。先跑通，再调优，最后把每次实验记录下来——这个顺序别反。希望帮到你。
本文还有配套的精品资源，点击获取
