尧图网站建设 
简介这份资源是基于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.csvuserId,movieId,rating,timestamp和 movies.csvmovieId,title,genres通过hdfs dfs -put落到 HDFS 的/movie/raw/目录。第二段是清洗与切分过滤掉评分次数少于阈值的冷门电影和僵尸用户把数据按时间切训练集和测试集。第三段是核心计算用 ItemCF 或 UserCF 算相似度再对每个用户生成候选推荐。第四段是结果落盘与评估推荐列表写到/movie/output/用 RMSE 或 PrecisionK 评估。这里有个关键设计决策相似度计算用 ItemCF 还是 UserCF。电影场景我一般选 ItemCF因为电影数量远小于用户数量物品相似度矩阵更小、更稳定而且「看了 A 的人也看了 B」这个解释对用户更直观。UserCF 在用户兴趣漂移快的场景更好但电影偏好相对稳定ItemCF 的性价比更高。数据流里每一步都要考虑分区。比如算物品相似度时如果按 movieId 做 key同一个物品的所有评分会进同一个 reducer但热门电影比如《肖申克的救赎》的评分可能有几十万条会造成数据倾斜。常见做法是给热门物品的 key 加随机后缀打散算完再合并这个后面避坑章节会细讲。3. 从零把 Hadoop 伪分布式跑起来安装、配置与验证3.1 环境准备与 JDK、Hadoop 安装先明确版本组合这是最容易翻车的地方。我一般用 JDK 8 配 Hadoop 3.3.xJDK 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 namefs.defaultFS/name valuehdfs://localhost:9000/value /property property namehadoop.tmp.dir/name value/opt/hadoop/tmp/value /property /configurationfs.defaultFS的端口 9000 是 NameNode 的 RPC 端口别和 Web UI 的 9870 搞混。hadoop.tmp.dir一定要显式指定默认在 /tmp 下机器重启就没了NameNode 元数据丢失会导致集群起不来。hdfs-site.xml设副本数为 1伪分布式只有一个 DataNodeconfiguration property namedfs.replication/name value1/value /property property namedfs.namenode.name.dir/name value/opt/hadoop/data/namenode/value /property property namedfs.datanode.data.dir/name value/opt/hadoop/data/datanode/value /property /configurationmapred-site.xml指定用 YARN 跑 MapReduceconfiguration property namemapreduce.framework.name/name valueyarn/value /property /configurationyarn-site.xml配 ResourceManager 和 NodeManagerconfiguration property nameyarn.resourcemanager.hostname/name valuelocalhost/value /property property nameyarn.nodemanager.aux-services/name valuemapreduce_shuffle/value /property /configurationyarn.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, movieIdReducer 收到一个用户看过的所有电影列表两两组合输出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 MapperLongWritable, 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 为 keymovieId 为 value 输出 context.write(new Text(userId), new Text(movieId)); } }Mapper 只做转发真正的组合逻辑在 Reducer因为同一个用户的所有电影必须进同一个 reducer 才能两两配对public class CoOccurrenceReducer extends ReducerText, Text, Text, IntWritable { Override protected void reduce(Text key, IterableText values, Context context) throws IOException, InterruptedException { ListString 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 ReducerText, IntWritable, Text, Text { private MapString, 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, IterableIntWritable 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到_9Reducer 端先局部聚合再二次聚合。或者在 Mapper 阶段就用context.getCounter统计每个 key 的量超过阈值的直接拆分。5.2 NameNode 反复格式化后 DataNode 起不来现象jps里只有 NameNode 没有 DataNode日志报clusterID mismatch。原因多次执行hdfs namenode -formatNameNode 的 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 和 PrecisionK。RMSE 衡量评分预测误差把测试集里的真实评分和预测评分算均方根误差值越小越好MovieLens 上 ItemCF 一般能到 0.85 到 0.95 之间。PrecisionK 衡量 TopN 推荐里有多少是用户真正喜欢的K 取 10 时能到 0.15 到 0.25 就算不错。但这两个指标都有盲区。RMSE 对推荐列表的排序不敏感PrecisionK 又依赖你怎么定义「喜欢」评分大于 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(fRecall10: {hit / total:.4f})这段脚本的关键参数是评分阈值 4.0 和 K10阈值调高召回率会降但精度升按业务场景定。跑完如果 Recall10 低于 0.1先别急着调算法回去看数据清洗是不是把太多有效评分过滤掉了。最后说个我自己的习惯每次改完相似度公式或参数一定先在小数据集比如 ml-latest-small上跑一遍全流程确认指标没崩再上大数据集。大数据集跑一次动辄半小时拿它调参是跟自己过不去。这套系统值不值得做取决于你是要交作业还是真要用交作业跑通伪分布式加 ItemCF 就够了真要用还得补实时召回和冷启动。希望帮到你。本文还有配套的精品资源点击获取 基于Hadoop电影推荐系统：从伪分布式搭建到ItemCF算法实现
简介这份资源是基于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.csvuserId,movieId,rating,timestamp和 movies.csvmovieId,title,genres通过hdfs dfs -put落到 HDFS 的/movie/raw/目录。第二段是清洗与切分过滤掉评分次数少于阈值的冷门电影和僵尸用户把数据按时间切训练集和测试集。第三段是核心计算用 ItemCF 或 UserCF 算相似度再对每个用户生成候选推荐。第四段是结果落盘与评估推荐列表写到/movie/output/用 RMSE 或 PrecisionK 评估。这里有个关键设计决策相似度计算用 ItemCF 还是 UserCF。电影场景我一般选 ItemCF因为电影数量远小于用户数量物品相似度矩阵更小、更稳定而且「看了 A 的人也看了 B」这个解释对用户更直观。UserCF 在用户兴趣漂移快的场景更好但电影偏好相对稳定ItemCF 的性价比更高。数据流里每一步都要考虑分区。比如算物品相似度时如果按 movieId 做 key同一个物品的所有评分会进同一个 reducer但热门电影比如《肖申克的救赎》的评分可能有几十万条会造成数据倾斜。常见做法是给热门物品的 key 加随机后缀打散算完再合并这个后面避坑章节会细讲。3. 从零把 Hadoop 伪分布式跑起来安装、配置与验证3.1 环境准备与 JDK、Hadoop 安装先明确版本组合这是最容易翻车的地方。我一般用 JDK 8 配 Hadoop 3.3.xJDK 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 namefs.defaultFS/name valuehdfs://localhost:9000/value /property property namehadoop.tmp.dir/name value/opt/hadoop/tmp/value /property /configurationfs.defaultFS的端口 9000 是 NameNode 的 RPC 端口别和 Web UI 的 9870 搞混。hadoop.tmp.dir一定要显式指定默认在 /tmp 下机器重启就没了NameNode 元数据丢失会导致集群起不来。hdfs-site.xml设副本数为 1伪分布式只有一个 DataNodeconfiguration property namedfs.replication/name value1/value /property property namedfs.namenode.name.dir/name value/opt/hadoop/data/namenode/value /property property namedfs.datanode.data.dir/name value/opt/hadoop/data/datanode/value /property /configurationmapred-site.xml指定用 YARN 跑 MapReduceconfiguration property namemapreduce.framework.name/name valueyarn/value /property /configurationyarn-site.xml配 ResourceManager 和 NodeManagerconfiguration property nameyarn.resourcemanager.hostname/name valuelocalhost/value /property property nameyarn.nodemanager.aux-services/name valuemapreduce_shuffle/value /property /configurationyarn.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, movieIdReducer 收到一个用户看过的所有电影列表两两组合输出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 MapperLongWritable, 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 为 keymovieId 为 value 输出 context.write(new Text(userId), new Text(movieId)); } }Mapper 只做转发真正的组合逻辑在 Reducer因为同一个用户的所有电影必须进同一个 reducer 才能两两配对public class CoOccurrenceReducer extends ReducerText, Text, Text, IntWritable { Override protected void reduce(Text key, IterableText values, Context context) throws IOException, InterruptedException { ListString 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 ReducerText, IntWritable, Text, Text { private MapString, 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, IterableIntWritable 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到_9Reducer 端先局部聚合再二次聚合。或者在 Mapper 阶段就用context.getCounter统计每个 key 的量超过阈值的直接拆分。5.2 NameNode 反复格式化后 DataNode 起不来现象jps里只有 NameNode 没有 DataNode日志报clusterID mismatch。原因多次执行hdfs namenode -formatNameNode 的 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 和 PrecisionK。RMSE 衡量评分预测误差把测试集里的真实评分和预测评分算均方根误差值越小越好MovieLens 上 ItemCF 一般能到 0.85 到 0.95 之间。PrecisionK 衡量 TopN 推荐里有多少是用户真正喜欢的K 取 10 时能到 0.15 到 0.25 就算不错。但这两个指标都有盲区。RMSE 对推荐列表的排序不敏感PrecisionK 又依赖你怎么定义「喜欢」评分大于 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(fRecall10: {hit / total:.4f})这段脚本的关键参数是评分阈值 4.0 和 K10阈值调高召回率会降但精度升按业务场景定。跑完如果 Recall10 低于 0.1先别急着调算法回去看数据清洗是不是把太多有效评分过滤掉了。最后说个我自己的习惯每次改完相似度公式或参数一定先在小数据集比如 ml-latest-small上跑一遍全流程确认指标没崩再上大数据集。大数据集跑一次动辄半小时拿它调参是跟自己过不去。这套系统值不值得做取决于你是要交作业还是真要用交作业跑通伪分布式加 ItemCF 就够了真要用还得补实时召回和冷启动。希望帮到你。本文还有配套的精品资源点击获取