简介：这份资源是基于Hadoop框架实现的电影推荐系统完整项目源码包，面向具备Java与大数据基础、希望实践分布式推荐算法的开发者与学习者。项目以HDFS与MapReduce为核心，结合Java实现数据收集、清洗、相似度计算与推荐生成等环节，并可能涉及协同过滤、内容过滤等推荐策略，适合作为大数据课程设计或毕业设计的参考案例。压缩包共1117个文件，约40.21MB，包含379个php、169个html、157个png、122个js、60个css等前端与页面资源，以及22个py脚本、8个docx文档、5个sql与若干xml、json配置文件，覆盖源码、静态资源与说明文档。目前已有279人学习下载。项目目录结构完整，包含推荐算法实现、Hadoop任务代码与前端展示模块，可帮助读者理解从数据预处理到推荐结果存储的全流程，并参考成熟项目的组织方式与排错思路。
1. 从一份「基于 Hadoop 电影推荐系统.zip」说起：它到底解决什么问题
如果你手里正好拿到一个叫「基于 hadoop 电影推荐系统.zip」的压缩包，第一反应大概率是：这东西能不能跑起来、跑起来之后推荐结果靠不靠谱、我照着搭一套要花多久。它本质上是一个用 Hadoop 生态做离线批处理、基于用户历史评分预测电影偏好的系统，核心链路是「评分数据落 HDFS → MapReduce/Spark 算相似度 → 产出 TopN 推荐列表」。它解决的不是「实时猜你想看」，而是「在千万级评分上稳定跑出可解释的推荐结果」，适合课程设计、毕设、以及想入门分布式计算又不想只写 WordCount 的工程师。下面我按自己搭过几套的经验，把选型、伪分布式搭建、算法实现、踩坑和验证一条线讲清楚，新手能跟着敲，熟手能直接看参数边界。
2. 为什么电影推荐要挂在 Hadoop 上：选型理由与数据流拆解
2.1 单机 pandas 能做的事，为什么非要上 Hadoop
很多人第一反应是：MovieLens 的 ml-latest-small 才 10 万条评分，pandas 两秒就算完了，上 Hadoop 是不是杀鸡用牛刀。这个判断在小数据集上没错，但「基于 hadoop 电影推荐系统」这个标题真正的价值在于数据规模上来之后的横向扩展能力。当评分表从 10 万涨到 2000 万、用户数到百万级，协同过滤里最贵的两步——用户-物品相似度矩阵和 TopN 排序——在单机上会直接吃爆内存。Hadoop 的解法是把评分按 userID 或 itemID 做 shuffle，让每个 reducer 只处理一部分键，内存压力被切碎到集群节点上。
另一个常被忽略的点是可复现性。单机脚本换个环境、换个 pandas 版本，groupby 的默认排序都可能变，结果对不上。MapReduce 的 shuffle 和 sort 语义是框架保证的，同样的输入、同样的分区函数，输出稳定。做课程设计要写报告、要答辩演示，这种确定性比省那几秒重要得多。
所以选型结论是：数据量在百万条评分以下、只求跑通，单机完全够；一旦你要演示「分布式」这个卖点，或者数据真的到了千万级，Hadoop 的 HDFS 加 MapReduce 就是最稳的底座。常见做法是底层用 HDFS 存原始 ratings，中间用 MapReduce 或 Spark 算，最后把推荐结果写回 HDFS 再用脚本导出成 CSV 给前端。
2.2 一条完整的离线推荐数据流长什么样
把整个系统拆开，数据流是四段。第一段是原始数据入湖：ratings.csv（userId,movieId,rating,timestamp）和 movies.csv（movieId,title,genres）通过hdfs dfs -put落到 HDFS 的/movie/raw/目录。第二段是清洗与切分：过滤掉评分次数少于阈值的冷门电影和僵尸用户，把数据按时间切训练集和测试集。第三段是核心计算：用 ItemCF 或 UserCF 算相似度，再对每个用户生成候选推荐。第四段是结果落盘与评估：推荐列表写到/movie/output/，用 RMSE 或 Precision@K 评估。
这里有个关键设计决策：相似度计算用 ItemCF 还是 UserCF。电影场景我一般选 ItemCF，因为电影数量远小于用户数量，物品相似度矩阵更小、更稳定，而且「看了 A 的人也看了 B」这个解释对用户更直观。UserCF 在用户兴趣漂移快的场景更好，但电影偏好相对稳定，ItemCF 的性价比更高。
数据流里每一步都要考虑分区。比如算物品相似度时，如果按 movieId 做 key，同一个物品的所有评分会进同一个 reducer，但热门电影（比如《肖申克的救赎》）的评分可能有几十万条，会造成数据倾斜。常见做法是给热门物品的 key 加随机后缀打散，算完再合并，这个后面避坑章节会细讲。
3. 从零把 Hadoop 伪分布式跑起来：安装、配置与验证
3.1 环境准备与 JDK、Hadoop 安装
先明确版本组合，这是最容易翻车的地方。我一般用 JDK 8 配 Hadoop 3.3.x，JDK 11 也能跑但部分脚本有兼容告警。操作系统用 Ubuntu 20.04 或 CentOS 7 都行，Windows 下建议直接用 WSL2，别在原生 Windows 上折腾，路径和权限问题会让你怀疑人生。
第一步装 JDK 并配环境变量：
# 安装 OpenJDK 8 sudo apt update sudo apt install -y openjdk-8-jdk # 验证版本，输出应包含 1.8.0 java -version # 配置 JAVA_HOME，写入 ~/.bashrc echo 'export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64' >> ~/.bashrc echo 'export PATH=$JAVA_HOME/bin:$PATH' >> ~/.bashrc source ~/.bashrcJAVA_HOME必须指向 JDK 根目录而不是 bin 目录，Hadoop 启动脚本会拼$JAVA_HOME/bin/java，指错了会报JAVA_HOME is not set。装完 JDK 再解压 Hadoop：
# 下载并解压到 /opt tar -zxvf hadoop-3.3.6.tar.gz -C /opt/ mv /opt/hadoop-3.3.6 /opt/hadoop # 配置 Hadoop 自身环境变量 echo 'export HADOOP_HOME=/opt/hadoop' >> ~/.bashrc echo 'export PATH=$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$PATH' >> ~/.bashrc source ~/.bashrc3.2 四个核心配置文件怎么改
伪分布式要改core-site.xml、hdfs-site.xml、mapred-site.xml、yarn-site.xml四个文件，都在$HADOOP_HOME/etc/hadoop/下。逐个说关键参数。
core-site.xml指定默认文件系统和临时目录：
<configuration> <property> <name>fs.defaultFS</name> <value>hdfs://localhost:9000</value> </property> <property> <name>hadoop.tmp.dir</name> <value>/opt/hadoop/tmp</value> </property> </configuration>fs.defaultFS的端口 9000 是 NameNode 的 RPC 端口，别和 Web UI 的 9870 搞混。hadoop.tmp.dir一定要显式指定，默认在 /tmp 下，机器重启就没了，NameNode 元数据丢失会导致集群起不来。
hdfs-site.xml设副本数为 1，伪分布式只有一个 DataNode：
<configuration> <property> <name>dfs.replication</name> <value>1</value> </property> <property> <name>dfs.namenode.name.dir</name> <value>/opt/hadoop/data/namenode</value> </property> <property> <name>dfs.datanode.data.dir</name> <value>/opt/hadoop/data/datanode</value> </property> </configuration>mapred-site.xml指定用 YARN 跑 MapReduce：
<configuration> <property> <name>mapreduce.framework.name</name> <value>yarn</value> </property> </configuration>yarn-site.xml配 ResourceManager 和 NodeManager：
<configuration> <property> <name>yarn.resourcemanager.hostname</name> <value>localhost</value> </property> <property> <name>yarn.nodemanager.aux-services</name> <value>mapreduce_shuffle</value> </property> </configuration>yarn.nodemanager.aux-services必须是mapreduce_shuffle，写错了 MapReduce 任务会卡在 shuffle 阶段不动。
3.3 格式化、启动与三个验证动作
配置改完，先格式化 NameNode，再启动：
# 格式化，只能执行一次，重复执行会清空元数据 hdfs namenode -format # 启动 HDFS 和 YARN start-dfs.sh start-yarn.sh # 验证进程，应看到 NameNode、DataNode、ResourceManager、NodeManager jpsjps输出里如果少了 DataNode，八成是hadoop.tmp.dir或dfs.datanode.data.dir权限不对，或者多次 format 导致 clusterID 不一致。验证 HDFS 能读写：
# 建目录并上传测试文件 hdfs dfs -mkdir -p /movie/raw echo "1,1,5.0" > test.csv hdfs dfs -put test.csv /movie/raw/ # 查看文件 hdfs dfs -ls /movie/raw/ hdfs dfs -cat /movie/raw/test.csv浏览器打开http://localhost:9870能看到 NameNode 页面、http://localhost:8088能看到 YARN 页面，说明集群健康。这三个验证动作做完，底座就算稳了。
4. 用 MapReduce 实现 ItemCF：相似度计算与 TopN 推荐
4.1 ItemCF 的两阶段 MapReduce 设计
ItemCF 的核心是「共现矩阵」：对每个用户看过的电影两两配对，统计共同观看次数，再除以各自流行度的归一化因子得到相似度。用 MapReduce 实现要拆成两个 Job。
Job1 算物品共现。Mapper 读一行评分，以 userId 为 key 输出<userId, movieId>，Reducer 收到一个用户看过的所有电影列表，两两组合输出<movieA:movieB, 1>。再一个 Reducer 聚合得到共现次数。
Job2 算相似度并生成推荐。Mapper 读共现结果，以 movieA 为 key 输出，Reducer 拿到 movieA 的所有共现对，除以归一化因子得到相似度，再结合用户历史评分加权，输出 TopN。
这个设计里最贵的是 Job1 的两两组合，一个用户看了 100 部电影就是 4950 对，所以要先过滤掉看电影超过 500 部的重度用户，否则单个 reducer 会被打爆。
4.2 共现矩阵的 Mapper 与 Reducer 代码
先看 Job1 的 Mapper，把用户-电影对转成电影两两组合：
public class CoOccurrenceMapper extends Mapper<LongWritable, Text, Text, IntWritable> { private Text pairKey = new Text(); private final static IntWritable ONE = new IntWritable(1); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { // 输入格式: userId,movieId,rating,timestamp String[] fields = value.toString().split(","); if (fields.length < 3) return; // 跳过脏数据 String userId = fields[0]; String movieId = fields[1]; // 以 userId 为 key，movieId 为 value 输出 context.write(new Text(userId), new Text(movieId)); } }Mapper 只做转发，真正的组合逻辑在 Reducer，因为同一个用户的所有电影必须进同一个 reducer 才能两两配对：
public class CoOccurrenceReducer extends Reducer<Text, Text, Text, IntWritable> { @Override protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { List<String> movies = new ArrayList<>(); for (Text v : values) { movies.add(v.toString()); } // 两两组合，输出 movieA:movieB -> 1 for (int i = 0; i < movies.size(); i++) { for (int j = i + 1; j < movies.size(); j++) { String a = movies.get(i); String b = movies.get(j); context.write(new Text(a + ":" + b), new IntWritable(1)); context.write(new Text(b + ":" + a), new IntWritable(1)); } } } }这里输出双向对是为了后面相似度矩阵对称，省得再转置。参数上要注意movies.size()如果超过几千，这个双重循环会非常慢，所以前面说的过滤重度用户必须在 Mapper 之前用一步清洗 Job 做掉。
4.3 相似度归一化与 TopN 推荐的实现
Job2 的 Reducer 拿到<movieA:movieB, 共现次数>，需要除以sqrt(N_A * N_B)做余弦归一化，其中 N_A 是电影 A 的观看人数。这个 N_A 可以在 setup 阶段从分布式缓存读入，或者再起一个 Job 算好放 HDFS。
public class SimilarityReducer extends Reducer<Text, IntWritable, Text, Text> { private Map<String, Integer> moviePopularity = new HashMap<>(); @Override protected void setup(Context context) throws IOException { // 从分布式缓存读取电影流行度文件 movieId\tcount URI[] cacheFiles = context.getCacheFiles(); if (cacheFiles != null) { for (URI uri : cacheFiles) { BufferedReader br = new BufferedReader(new InputStreamReader( new FileInputStream(uri.getPath()))); String line; while ((line = br.readLine()) != null) { String[] parts = line.split("\t"); moviePopularity.put(parts[0], Integer.parseInt(parts[1])); } br.close(); } } } @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { String[] pair = key.toString().split(":"); String movieA = pair[0]; String movieB = pair[1]; int coCount = 0; for (IntWritable v : values) { coCount += v.get(); } Integer popA = moviePopularity.get(movieA); Integer popB = moviePopularity.get(movieB); if (popA == null || popB == null || popA == 0 || popB == 0) return; // 余弦相似度 double similarity = coCount / Math.sqrt((double) popA * popB); context.write(new Text(movieA), new Text(movieB + ":" + similarity)); } }setup里读分布式缓存是标准做法，用context.getCacheFiles()拿 URI，提交 Job 时用job.addCacheFile()注册。相似度算完，最后一步是给每个用户生成推荐：拿用户看过的电影，查相似度矩阵，加权求和排序取 TopN。这一步可以再写一个 Job，也可以把相似度矩阵加载到内存用单机脚本做，取决于矩阵大小。百万级电影的话矩阵太大，还是得用 MapReduce。
5. 这套系统最容易翻车的五个地方：排查与避坑
5.1 数据倾斜导致某个 Reducer 卡在 99%
现象：Job 跑到 99% 不动，看 YARN 页面发现某个 reducer 处理的数据量是其他的几十倍。原因：热门电影或活跃用户作为 key，所有相关记录都进同一个 reducer。解决：对热门 key 加随机后缀打散，比如给《肖申克的救赎》的 key 拼上_0到_9，Reducer 端先局部聚合再二次聚合。或者在 Mapper 阶段就用context.getCounter统计每个 key 的量，超过阈值的直接拆分。
5.2 NameNode 反复格式化后 DataNode 起不来
现象：jps里只有 NameNode 没有 DataNode，日志报clusterID mismatch。原因：多次执行hdfs namenode -format，NameNode 的 clusterID 变了，但 DataNode 的 VERSION 文件还是旧的。解决：停掉集群，删掉dfs.datanode.data.dir下的所有内容，重新hdfs namenode -format一次，再启动。血泪经验是 format 之前一定确认没有重要数据，这个操作没有后悔药。
5.3 内存不足导致 Container 被 Kill
现象：任务报Container killed on request. Exit code is 137。原因：Mapper 或 Reducer 的堆内存不够，或者 YARN 的yarn.nodemanager.resource.memory-mb设得太小。解决：在mapred-site.xml里调大mapreduce.map.memory.mb和mapreduce.reduce.memory.mb，同时确认yarn.nodemanager.resource.memory-mb大于两者之和。伪分布式单机内存有限，建议 map 给 1024MB、reduce 给 2048MB 起步。
5.4 中文电影名乱码
现象：推荐结果里电影名显示成问号或方块。原因：原始 CSV 是 UTF-8，但 MapReduce 默认按平台编码读，或者输出时没指定编码。解决：在 Job 里显式设置job.getConfiguration().set("mapreduce.output.textoutputformat.separator", ",")，读写文件时统一用StandardCharsets.UTF_8，别依赖系统默认。
5.5 推荐结果全是热门电影
现象：每个用户的 TopN 推荐几乎一样，都是那几部高分大片。原因：相似度没做流行度惩罚，热门电影和谁都共现高。解决：在相似度公式里加1 / log(1 + popularity)做惩罚，或者用 ItemCF 的改进版归一化。这个坑很隐蔽，因为 RMSE 指标可能看着还行，但推荐多样性极差，答辩时容易被问住。
6. 怎么验证推荐质量：离线指标与一个我常用的抽样技巧
系统跑通只是第一步，能证明推荐有效才算落地。离线评估最常用的是 RMSE 和 Precision@K。RMSE 衡量评分预测误差，把测试集里的真实评分和预测评分算均方根误差，值越小越好，MovieLens 上 ItemCF 一般能到 0.85 到 0.95 之间。Precision@K 衡量 TopN 推荐里有多少是用户真正喜欢的，K 取 10 时能到 0.15 到 0.25 就算不错。
但这两个指标都有盲区。RMSE 对推荐列表的排序不敏感，Precision@K 又依赖你怎么定义「喜欢」（评分大于 3.5 还是 4.0）。我一般会加一个抽样人工检查：随机抽 20 个用户，把他们最近看过的 5 部电影从训练集里拿掉，看系统能不能在 TopN 里推回来。这个技巧比任何指标都直观，答辩时演示这个比念 RMSE 有说服力。
具体做法是写一个 Python 脚本读 HDFS 导出的推荐结果和原始评分：
import pandas as pd # 读推荐结果和测试集 recs = pd.read_csv('recommendations.csv') # userId,movieId,score test = pd.read_csv('test_ratings.csv') # userId,movieId,rating # 只看评分 >= 4.0 的作为正样本 positive = test[test['rating'] >= 4.0] # 对每个用户算命中率 hit = 0 total = 0 for uid, group in positive.groupby('userId'): true_movies = set(group['movieId']) rec_movies = set(recs[recs['userId'] == uid].head(10)['movieId']) hit += len(true_movies & rec_movies) total += len(true_movies) print(f'Recall@10: {hit / total:.4f}')这段脚本的关键参数是评分阈值 4.0 和 K=10，阈值调高召回率会降但精度升，按业务场景定。跑完如果 Recall@10 低于 0.1，先别急着调算法，回去看数据清洗是不是把太多有效评分过滤掉了。
最后说个我自己的习惯：每次改完相似度公式或参数，一定先在小数据集（比如 ml-latest-small）上跑一遍全流程，确认指标没崩再上大数据集。大数据集跑一次动辄半小时，拿它调参是跟自己过不去。这套系统值不值得做，取决于你是要交作业还是真要用，交作业跑通伪分布式加 ItemCF 就够了，真要用还得补实时召回和冷启动。希望帮到你。
本文还有配套的精品资源，点击获取
