简介：基于Hadoop的电影推荐系统设计与实现方案，适合大数据、计算机相关专业学生作为小组作业、课程设计或毕业设计参考，也可供初学者了解推荐系统与Hadoop生态的结合方式。方案围绕电影评分数据，实现基于用户或物品的协同过滤推荐流程，涵盖数据存储、处理、算法调用与结果展示等环节。资源共9个文件，压缩包仅100KB，以xml、properties等配置文件为主，覆盖Spring、Struts框架的集成、数据库连接池配置以及日志、C3P0等基础设施；同时包含核心jar包与项目结构文件，便于直接导入IDE查看运行。已有479人学习下载。该方案代码经测试通过，答辩评分较高，可作为完整项目参考；下载后可与作者联系获取远程教学支持，特别适合需要快速搭建推荐系统原型、但基础相对薄弱的同学。
1. 小组大作业里，Hadoop电影推荐系统究竟要“设计”什么
如果你的小组领到“基于Hadoop的电影推荐系统设计与实现”这个题目，最该明确的不是推荐算法能跑多准，而是这门课要考察的“大数据处理链条”是否完整。老师想看到的是原始评分数据进入 HDFS，经过清洗、倒排索引、相似度计算，最终输出每个用户的 Top-N 推荐列表，每一步都落在 Hadoop 生态上。所以这篇博文直接把课程设计拆成四块：算法选型、环境搭建、MapReduce 数据管道、答辩高频参数。新手可以照着把第一版跑通，带过课设的人也能借机把集群部署和调参说圆。
2. 电影推荐系统的算法选型：为什么基于物品的协同过滤更适合 MapReduce
2.1 三种推荐算法在 Hadoop 作业里的实现成本
课程设计最怕选一个算法听起来高级，结果用 MapReduce 写不出。下表把三种常见算法的实现成本摆出来，再决定选哪个能按期交作业。
| 算法 | 核心思想 | 纯 MapReduce 可算性 | 作业展示效果 |
|---|---|---|---|
| 基于用户的 CF | 找相似用户，推荐相似用户看过的物品 | 用户数远超物品数，相似度矩阵爆炸 | 直觉好懂，但不想算 |
| 基于物品的 CF | 找相似物品，推荐用户历史物品的相似物 | 物品数通常远小于用户数，适合层层 MapReduce | 最容易拆解并画数据流图 |
| ALS 矩阵分解 | 隐因子模型，交替最小二乘 | 每次迭代都要多轮 MR，或直接换 Spark | 效果最好，但课设答辩难以说清每轮 shuffle |
我一般建议小组成员选基于物品的协同过滤。原因有两个：一是电影数据集中用户量往往几十万，物品量只有几千到几万，物-物相似度矩阵能落盘也能放进内存；二是算法可以被拆成三张有先后依赖的 MapReduce 作业表，正好对应“设计文档”里的数据流图。ALS 再诱人也只需要在报告里提一句“可作为扩展方向”。
2.2 用三步 MapReduce 作业链实现物品协同过滤
基于物品的 CF 的核心公式是余弦相似度：
sim(i, j) = (所有同时给 i 和 j 评分的用户，其评分乘积之和) / (sqrt(i 的评分平方和) * sqrt(j 的评分平方和))这个公式可以拆成三个 Hadoop 作业：
- J1：从原始评分表生成“每个用户的历史评分列表”，输出格式
userId \t item1:score1,item2:score2。 - J2：读取 J1 输出，对同一个用户下的物品做两两组合，统计物品对共现次数并计算相似度。
- J3：用相似度矩阵给每个用户未看过的物品累加预测评分，输出 Top-N。
J1 的 Mapper 和 Reducer 用 Hadoop Streaming 写 Python 就可以，下面是能直接跑的代码：
#!/usr/bin/env python3 # map_user_item.py import sys for line in sys.stdin: line = line.strip() if line.startswith("userId") or not line: continue uid, mid, score, _ = line.split(",") print(f"{uid}\t{mid}:{score}")#!/usr/bin/env python3 # reduce_user_item.py import sys current_user = None item_scores = [] def flush(user, items): # 把同一用户的所有物品聚成一行，以逗号分隔 print(f"{user}\t{','.join(items)}") for line in sys.stdin: uid, items = line.strip().split("\t") if uid != current_user: if current_user is not None: flush(current_user, item_scores) current_user = uid item_scores = [] item_scores.append(items) if current_user is not None: flush(current_user, item_scores)这段代码的作用是把稀疏评分表变成稠密的一行式用户画像。map_user_item.py只做格式转换，reduce_user_item.py则依赖 Hadoop 的 shuffle 机制保证同一个 userId 的记录进入同一个 Reducer。参数方面，Hadoop Streaming 默认以\t作为 key/value 分隔符，命令行中不需要额外声明，但如果评分字段中有空格或特殊字符，建议在脚本里先 strip。
运行 J1 的命令：
hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-3.3.6.jar \ -files map_user_item.py,reduce_user_item.py \ -mapper "python3 map_user_item.py" \ -reducer "python3 reduce_user_item.py" \ -input /data/movielens/ratings.csv \ -output /data/user_items把3.3.6换成你集群里的实际 Hadoop 版本。-files会把脚本分发到所有 NodeManager 的工作目录，-input和-output分别是 HDFS 路径，输出目录不能已存在，否则作业会失败。
2.3 去均值与降权：两个值得写进报告里的细节
纯余弦相似度有个问题：有的用户习惯全打 4 分，有的用户全打 2 分，导致两个在“口味偏好”上一致的用户因评分习惯不同而相似度偏低。在课程设计的答辩里，说出“对用户评分做均值中心化”会是一个明显的加分点，具体做法是：在 J1 输出前，先用一次 MR 统计每个用户的平均分，然后对每条评分做rating - avg_rating。
另一个更工程化的细节是热门物品降权。如果《肖申克的救赎》被 90% 的用户评分，那么它和任何物品的相似度都会虚高，导致推荐结果全是热门电影，看不出个性化。常见做法是在 J2 的相似度公式里乘以一个1 / log(1 + item_freq)系数，这个系数可以提前统计好放进分布式缓存。
这两个 point 都不是必须的，但写进设计文档能让“大作业”看起来像系统设计，而不是单纯的调包。
3. Hadoop 环境搭建：从伪分布式到集群部署策略的取舍
3.1 伪分布式搭建的关键配置
大作业规模的数据量通常只有几 MB 到几十 MB，完全不需要一上来就搭集群。先用伪分布式跑通逻辑，再按组员考勤或者答辩要求决定是否升级集群。
伪分布式的标准路径是下载 Hadoop 二进制包、配置JAVA_HOME、修改四个 XML 文件。最核心的是core-site.xml和hdfs-site.xml：
<!-- core-site.xml --> <configuration> <property> <name>fs.defaultFS</name> <value>hdfs://localhost:9000</value> </property> <property> <name>hadoop.tmp.dir</name> <value>/home/hadoop/tmp</value> </property> </configuration><!-- hdfs-site.xml --> <configuration> <property> <name>dfs.replication</name> <value>1</value> </property> <property> <name>dfs.namenode.name.dir</name> <value>file:///home/hadoop/dfs/name</value> </property> <property> <name>dfs.datanode.data.dir</name> <value>file:///home/hadoop/dfs/data</value> </property> </configuration>fs.defaultFS指定了默认 HDFS 地址，伪分布式通常用 9000 端口；dfs.replication必须改成 1，因为只有一个 DataNode，如果保持默认 3，写入时会产生数据块副本等待超时。格式化 NameNode 只需要执行一次：
hdfs namenode -format start-dfs.sh jpsjps能看到NameNode、DataNode、SecondaryNameNode三个进程，说明 HDFS 启动正常。之后再用yarn resourcemanager和yarn nodemanager启动计算层，串起后面要跑的 MapReduce。
3.2 三节点集群的部署策略：角色别再搞混
伪分布式只能说明“HDFS 能写文件、YARN 能跑作业”，但答辩时经常会被问“你们为什么不用伪分布式？”这时需要拿出集群部署策略。三节点是小组课设最常见的规模，角色分配如下：
| 节点 | 部署进程 | 说明 |
|---|---|---|
| node01 | NameNode, SecondaryNameNode, ResourceManager | 主节点，负责元数据和资源调度 |
| node02 | DataNode, NodeManager | 存储数据并执行 map/reduce 任务 |
| node03 | DataNode, NodeManager | 存储数据并执行 map/reduce 任务 |
配置集群时最重要的是workers文件（旧版叫slaves），一行一个节点名：
node02 node03还有/etc/hosts的映射关系，例如192.168.1.10 node01。各节点之间用 SSH 免密登录，否则start-dfs.sh在远程启动 DataNode 时会卡在输入密码。做完这些基础，再把你刚从伪分布式里学会的 XML 配置复制到每台机器的相同目录，然后把dfs.replication改回 3。到这里，集群就算能用了。
3.3 与 Zookeeper 整合的 HA 配置要点
如果老师要求“高可用”，就必须把 Hadoop 与 Zookeeper 整合。所谓 HA（High Availability）就是给 NameNode 做热备：同一时刻只有一个 Active NameNode 对外服务，另一个 Standby 节点实时同步命名空间状态。Zookeeper 负责故障切换时的分布式锁和选主。
配置 HA 比伪分布式多两个动作：一是启动 Zookeeper 集群，二是在hdfs-site.xml里声明 Nameservice。下面是最小可用配置片段：
<property> <name>dfs.nameservices</name> <value>mycluster</value> </property> <property> <name>dfs.ha.namenodes.mycluster</name> <value>nn1,nn2</value> </property>接着要配置 journalnode，让两个 NameNode 通过 JN 共享编辑日志。常见做法是在三台节点上各启动一个 JournalNode，而不是单独再开三台机器：
hdfs --daemon start journalnode这里有个很现实的坑：很多课设小组把 HA 部署写成“三台机器：一个主 NameNode、一个备 NameNode、一个 JournalNode”，这其实是错误的，因为 JournalNode 需要奇数个（通常是 3 个）才能完成日志写入定序。如果只有三个节点，最合理的是每个节点各跑一个 JournalNode，同时其中两个分别跑 Active/Standby NameNode。这个细节写进报告，会让答辩老师觉得你们真的动手敲过命令。
4. 电影推荐系统的数据管道实现：从 CSV 到 Top-N 列表
4.1 设计 HDFS 目录结构与输入格式
数据落进 HDFS 之前，先把目录规划好。我习惯在 HDFS 根目录下按“数据集/阶段”分层：
/data/movielens/ratings.csv /data/movielens/movies.csv /data/user_items/ /data/item_pairs/ /data/similarity/ /data/recommend//data/user_items是 J1 的输出，/data/item_pairs是 J2 的中间结果，/data/similarity是每个物品及其相似物品列表，/data/recommend是最终推荐列表。这样设计的好处是可以在 HDFS 每个输出目录下都用hdfs dfs -tail查看样例，方便在答辩时快速展示中间结果。
输入数据用 MovieLens 常见格式：userId,movieId,rating,timestamp，字段用逗号分隔。要提醒的是：Hadoop Streaming 默认按\t切分 key/value，如果直接拿 CSV 做 mapper 输入，需要自己在脚本里 split 逗号。我们接下来所有 mapper 都会显式做line.split(",")。
4.2 运行 J1：构建用户-物品评分列表
J1 的脚本已经在 2.2 节给出。核心目标是让某个用户的所有评分出现在同一行，这样 J2 的 Mapper 无需跨 Map 任务聚合就能直接生成物品对。如果漏掉这一步，直接让 Mapper 对原始评分做两两组合，同一个用户的数据分散在不同 Map 任务里，就会产生大量重复的物品对，而且无法用简单 Reducer 去重。
J1 作业跑完后，用以下命令检查输出：
hdfs dfs -cat /data/user_items/part-00000 | head预期看到类似输出：
1 296:5,101:4,321:3,...第 4.1 节设计的目录结构里，/data/user_items正是这个输出。如果发现同一用户出现多行，说明 J1 的 Reducer 没有正确按照 userId 分组，通常是默认分隔符问题，需要用-D stream.map.output.field.separator=\t显式指定。
4.3 运行 J2：计算物品共现与相似度
J2 的 Mapper 读取user \t item:score,item:score这种行，把同一用户下的所有物品两两组合。这里有一个计算细节：在组合前先过滤掉低分物品（比如只保留score >= 3的），能大幅减少输出量，而且推荐质量不一定下降，因为用户不喜欢的物品不该成为候选来源。
#!/usr/bin/env python3 # map_item_pair.py import sys for line in sys.stdin: uid, items_str = line.strip().split("\t") items = [] for entry in items_str.split(","): mid, score = entry.split(":") if float(score) >= 3.0: items.append(mid) for i in range(len(items)): for j in range(i + 1, len(items)): # 双端输出，保证相似度矩阵对称 print(f"{items[i]}\t{items[j]}\t1") print(f"{items[j]}\t{items[i]}\t1")这段代码的核心是用内存中的循环组合物品对，因为一个用户看过的电影数量有限（通常不超过几百），两层循环完全可承受。如果直接照搬这个逻辑到原始评分数据集上，每个用户的评分记录不会出现在同一行，因此必须先做 J1。
J2 的 Reducer 负责累加共现次数：
#!/usr/bin/env python3 # reduce_item_pair.py import sys current_pair = None count = 0 def flush(pair, cnt): i, j = pair.split("\t") print(f"{i}\t{j}\t{cnt}") for line in sys.stdin: item = line.strip().split("\t") pair = f"{item[0]}\t{item[1]}" if pair != current_pair: if current_pair is not None: flush(current_pair, count) current_pair = pair count = 0 count += int(item[2]) if current_pair is not None: flush(current_pair, count)J2 的输出是一张物品共现矩阵。如果 J2 的 Mapper 不做低分过滤，输出数量会爆炸：假设一个用户看了 500 部电影，就会生成 25 万个物品对，而一个课程数据集往往有几千个用户。因此低分过滤不是优化，而是“必须”。
4.4 生成最终推荐列表
共现次数还不能直接当相似度，需要按公式转化成 0-1 之间的余弦相似度。这一步可以用另一个 MapReduce 作业，但我更推荐把共现矩阵拉回本地做一次后处理，因为数据集不大时，调度一次 MR 的时间远大于计算本身。
在项目目录下执行：
hdfs dfs -getmerge /data/item_pairs cooccur.tsv python3 gen_similarity.py cooccur.tsv similarity.tsvgen_similarity.py的核心逻辑：
# gen_similarity.py import sys cooccur = {} freq = {} with open(sys.argv[1], "r") as f: for line in f: i, j, cnt = line.strip().split("\t") cnt = int(cnt) cooccur[(i, j)] = cnt freq[i] = freq.get(i, 0) + cnt # 近似的物品总评分次数，实际应统计每个物品独立出现次 with open(sys.argv[2], "w") as out: for (i, j), cnt in cooccur.items(): sim = cnt / (freq[i] * freq[j]) ** 0.5 out.write(f"{i}\t{j}\t{sim}\n")这个脚本里的freq用物品对的“总出现次数”近似物品 i 在数据集里的评分次数，严格讲应该用 J1 之前统计的物品独立频次，但课程设计用近似也能跑通。如果想要精确值，在 J2 之前先跑一个 WordCount 统计每个 item 的出现次数，把这批结果分不到分布式缓存里即可。
最后生成 Top-N 推荐时，对每个用户：读user_items中该用户的历史物品，把每个相似物品的相似度累加为得分，排序取出前 10 个没有看过的物品，写入/data/recommend。这一步用普通 Python 单机处理就行，不需要再起 MR 作业，答辩时把“流程选择”讲清楚，反而不是缺点。
5. 答辩前必被问的三个 MapReduce 参数与一个验证技巧
5.1 三个必调参数：块大小、副本数、Map 内存
小组答辩时，老师最喜欢指着mapred-site.xml问“你们为什么要改这里”。下表列出三个高频问题及其标准答法：
| 参数 | 默认值 | 课设推荐 | 原因 |
|---|---|---|---|
dfs.blocksize | 128MB | 64MB | 数据量小时，Block 过大导致 Map 任务数少，并行度上不去 |
dfs.replication | 3 | 集群 3，伪分布式 1 | 伪分布式只有 1 个 DataNode，设 3 会写入超时报错 |
mapreduce.map.memory.mb | 1024MB | 2048MB | J2 的 Mapper 在内存中生成物品对，1GB 容易触发 OOM |
调块大小不要在hdfs-site.xml里全局改，而是在运行作业时加参数：
hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-3.3.6.jar \ -D dfs.blocksize=67108864 \ -D mapreduce.map.memory.mb=2048 \ ...67108864就是 64MB。这样改的好处是只对当前作业生效，不会影响 HDFS 上其他数据块的副本策略和负载均衡。
5.2 用测试集快速验证推荐质量
推荐系统跑完不能只说“出来了”，要能给出一个数字。最简单的评估方法是把评分数据按用户随机分成训练集和测试集，80% 训练，20% 测试，然后用你的推荐结果和测试集中的真实观看记录计算 Precision@K 和 Recall@K。
# evaluate.py def evaluate(pred_dict, test_dict, k=10): hits = sum( len(set(pred_dict.get(uid, [])[:k]) & set(items)) for uid, items in test_dict.items() ) precision = hits / (len(test_dict) * k) recall = hits / sum(len(v) for v in test_dict.values()) return precision, recall注意pred_dict的 value 是每个用户的推荐电影 ID 列表，test_dict是用户在测试集中实际交互过的电影 ID。这个脚本不依赖 Hadoop，直接本地运行。但有一个坑：测试集里很多用户可能没有出现在训练集，需要先过滤掉冷启动用户。在评估前，建议打印一下测试集覆盖率和平均历史长度，写上“本实验排除了电影数少于 5 部的长尾用户”，这句话比任何指标都更能挡住答辩追问。
最后还有一个小技巧：调k的时候不要只调一个值，跑一遍for k in [5, 10, 20]，把结果做成三行表格放进报告。老师看到你不仅会跑通 pipeline，还知道评价指标对长度敏感，这组设计就能稳稳收尾。
本文还有配套的精品资源，点击获取
