简介针对大数据与推荐系统课程设计场景这份基于Hadoop的商品推荐系统项目包提供了完整的可参考实现适合高校学生、开发者快速上手分布式推荐系统搭建。包内共19个文件以Java源码和XML配置为主另含properties环境配置、README说明及txt文档合计26KB结构简洁便于直接导入工程进行二次开发。课程设计内容涵盖Hadoop核心组件HDFS与MapReduce的数据处理流程以及协同过滤、基于内容、混合推荐等主流算法项目源码按数据输入层、数据处理层、模型训练层和推荐服务层分模块组织清晰呈现从用户行为数据清洗到个性化推荐结果生成的完整链路。同时给出YARN资源调度、矩阵分解等优化思路与准确率、召回率等评估指标便于学生理解推荐系统工程的落地要点。已有3435人学习下载适合作为课程报告、毕业设计或企业入门实践的参考资料。1. 为什么课程设计选它Hadoop 与推荐系统天然互补拿到这份《基于 Hadoop 商品推荐系统课程设计》压缩包时我第一反应不是看代码而是先看工程结构——GoodRecommendationManagementSystem-master这种命名方式基本能判断是教学用的 Maven 项目。真正把它跑起来后会发现这套设计把 HDFS 存储、MapReduce 计算和协同过滤推荐串在了一条完整链路上而不是像很多 Demo 那样只写个单机推荐算法交差。课程的落点很明确让你理解推荐系统在电商场景下的数据流动同时把 Hadoop 分布式能力真的用起来。适合正在做大数据方向课程设计、或者想把 Hadoop 生态应用在具体业务上的同学也适合想回看经典 MapReduce 编程模型的工程师。从解压到跑通每一步都需要动手改配置、调参数这也是它的价值所在。2. 系统架构从用户点击到推荐结果的四层数据链路拿到项目后建议先不急着看 Mapper 和 Reducer而是对照README.md理清系统的分层。一个能跑通的 Hadoop 推荐系统必然不是把算法代码扔到集群上就行它需要对数据输入、存储、计算和输出做完整设计。2.1 四层架构与 Hadoop 组件对应关系这个项目的核心架构可以拆成四层每一层都能在 Hadoop 生态里找到对应组件下面是实际工程中最常见的映射关系层级职责Hadoop 组件典型数据形态数据输入层收集用户行为日志、商品信息写入分布式存储HDFS Client / Flume用户 ID、商品 ID、行为类型、时间戳数据处理层清洗、过滤、聚合原始数据生成推荐算法所需的中间表MapReduce / Hive去重后的行为记录、用户-商品评分矩阵模型训练层计算商品相似度、用户偏好向量生成推荐模型MapReduce 多轮作业商品共现矩阵、相似度 Top-K 列表推荐服务层接收查询请求根据模型匹配 Top-N 结果返回给前端HBase / Redis / 纯 Java API用户 ID 对应的推荐列表很多初学者会忽略一个关键点推荐服务层其实不在 Hadoop 的核心计算链路里它更多是读取模型结果。所以在这个课程设计中模型训练的产出物——比如商品相似度文件或用户推荐列表——需要落回 HDFS 或者导出到本地文件才能被上层应用使用。2.2 HDFS 目录规划与数据预处理存储项目的输入数据通常是一份用户行为日志字段格式类似userId,itemId,behaviorType,timestamp。这里有一个常见坑如果直接把原始日志丢给推荐算法结果会非常差因为日志中有大量浏览行为、重复点击、以及没有登录的游客 ID。所以必须先做数据清洗。我一般会在 HDFS 上建三个目录分别存放原始数据、清洗后数据、推荐模型输出# 创建 HDFS 目录依次存放原始日志、清洗结果、模型输出 hdfs dfs -mkdir -p /recommend/input hdfs dfs -mkdir -p /recommend/clean hdfs dfs -mkdir -p /recommend/output # 上传本地日志文件到 HDFS hdfs dfs -put user_behavior.log /recommend/input/这里说下为什么划分这三个目录input目录只做数据落盘不参与计算clean目录存储 MapReduce 清洗作业的输出供后续相似度计算作业读取output目录存放每轮作业的结果。如果所有作业的输入输出都指向同一个目录MapReduce 的OutputCommitter会直接报错这也是 Hadoop 编程中非常容易踩的坑。2.3 数据表结构与字段约束在写 MapReduce 之前你需要明确输入数据的字段顺序和类型。这个项目的数据格式虽然简单但强烈建议你在 README 里补一张字段约束表否则换一套数据就很容易出现ClassCastException。字段名类型示例约束说明user_idStringU12345不能为空游客 ID 需过滤item_idStringI789商品 ID需要与商品表关联behavior_typeStringclick / buy / collect不同行为对应不同权重timestampLong1612137600Unix 时间戳用于时间窗口过滤这里有一个设计决策为什么不直接使用原始的浏览记录我个人的做法是在清洗阶段做行为去重同一个用户在同一天对同一个商品的多次点击只保留一次但buy行为不合并。原因是buy行为的置信度远高于click去重会损失关键特征。3. 数据清洗与用户-商品矩阵构建写第一个生产级 MapReduce 作业这一章是整个项目中的基础也是 Hadoop 入门者对 MapReduce 理解最容易出偏差的地方。课程设计的核心不是调包调参而是能自己写一个从原始日志到用户-商品行为矩阵的完整作业。3.1 清洗逻辑设计什么数据该过滤在写 Mapper 之前先定义清洗规则。我建议至少做这四件事:过滤掉user_id为空的记录过滤掉behavior_type不在预设集合click、buy、collect中的记录对同一用户对同一商品的重复点击记录去重过滤掉时间戳超出业务范围的过期数据这些规则用 MapReduce 实现时map阶段负责解析和过滤reduce阶段负责按userId itemId做去重聚合。这样的设计能让 Map 端通过context.write输出的中间数据量显著减少减轻 Shuffle 的压力。3.2 Mapper 实现与参数说明下面是清洗作业的 Java Mapper 代码这个写法在 Hadoop 2/3 中都可以直接编译运行:import java.io.IOException; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; public class CleanMapper extends MapperLongWritable, Text, Text, Text { // 输出 key 使用 userId_itemIdvalue 使用行为类型加时间戳 private Text outKey new Text(); private Text outValue new Text(); Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line value.toString().trim(); if (line.isEmpty()) { return; } String[] fields line.split(,); // 字段数量不足或 user_id 为空直接丢弃 if (fields.length 4 || fields[0].isEmpty()) { return; } String userId fields[0]; String itemId fields[1]; String behavior fields[2]; String timestamp fields[3]; // 只保留已知的行为类型避免脏数据参与计算 if (!click.equals(behavior) !buy.equals(behavior) !collect.equals(behavior)) { return; } outKey.set(userId _ itemId); // 将行为和原始时间戳拼接作为 value outValue.set(behavior \t timestamp); context.write(outKey, outValue); } }这里关键参数是 Mapper 的泛型LongWritable, Text, Text, Text输入是行偏移量和文本输出 key 是用户商品组合键value 是行为与时间戳。为什么不用TextPair做组合键因为这里不需要在 Map 端排序后按用户分组直接用字符串拼接减少自定义 WritableComparable 的复杂度适合课程设计阶段的体量。3.3 Reducer 去重与生成行为向量Reducer 端的工作是处理同一个userId_itemId下的多条记录。由于 Map 输出已经按 key 分组Reducer 只需遍历 values保留权重最高的行为。比如buy的优先级大于collect大于click。import java.io.IOException; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; public class CleanReducer extends ReducerText, Text, Text, NullWritable { Override protected void reduce(Text key, IterableText values, Context context) throws IOException, InterruptedException { String bestBehavior click; // 默认最低优先级 long bestTimestamp 0L; for (Text val : values) { String[] parts val.toString().split(\t); if (parts.length ! 2) { continue; } String behavior parts[0]; long ts Long.parseLong(parts[1]); // 行为优先级判断buy collect click int currentPriority getPriority(behavior); if (currentPriority getPriority(bestBehavior) || (currentPriority getPriority(bestBehavior) ts bestTimestamp)) { bestBehavior behavior; bestTimestamp ts; } } // 输出格式userId_itemId behavior String[] keyParts key.toString().split(_); context.write(new Text(keyParts[0] \t keyParts[1] \t bestBehavior), NullWritable.get()); } private int getPriority(String behavior) { if (buy.equals(behavior)) return 3; if (collect.equals(behavior)) return 2; return 1; } }这里有两个值得注意的细节。第一Reducer 的 key 是Text类型但实际 value 中包含了下划线分隔的 userId 和 itemId所以输出时做了split(_)。第二NullWritable作为输出 value 表示我们不需要保留中间聚合值直接写最终文本。如果你后续要读入 Hive 表建议改成Text输出并加一个列分隔符比如\t。3.4 提交作业与观察运行状态写完 Maven 项目后需要把代码打成 jar 包并提交到集群。课程设计环境中通常是伪分布式直接用hadoop jar命令提交即可# 打包跳过测试避免 test 类干扰 mvn clean package -DskipTests # 提交清洗作业指定输入输出目录 hadoop jar target/recommend-system-1.0.jar CleanJob /recommend/input /recommend/clean # 查看输出结果 hdfs dfs -cat /recommend/clean/part-r-00000 | head -20命令中的CleanJob是包含main方法的类名需要在其中设置setMapperClass、setReducerClass、setOutputKeyClass等参数。这里我补充一个容易被忽略的参数job.setNumReduceTasks(2)。伪分布式默认只有一个 Reducer但如果你希望多个 Reduce 任务以加深对 Shuffle 的理解可以显式设置为 2。但要注意Reduce 任务数不是越多越好每个 Reducer 会分别输出一个part-r-xxxxx文件下游作业需要通配读取。4. 基于物品的协同过滤用 MapReduce 分步实现相似度计算与 Top-N 推荐课程设计的核心算法是协同过滤。很多人一上来就写一个cosineSimilarity的类但放到 Hadoop 上就必须拆解成多个 MapReduce 作业——因为相似度矩阵的规模可能远超单机内存。4.1 算法选型为什么用 Item-based CF 而不是 User-based CF在电商场景中物品数量通常远小于用户数量。以 10 万用户、1 万商品为例User-based CF 需要计算用户之间的相似度矩阵规模是 10 万 × 10 万而 Item-based CF 只需要计算 1 万 × 1 万。Hadoop 的 MapReduce 本质上更适合「先分割再合并」的计算Item-based 的每两两商品组合可以用一个 Reduce 任务处理天然符合分布式模型。因此这个课程设计选用 Item-based CF 是合理的选择。4.2 第一阶段构造商品共现矩阵共现矩阵的含义是在同一用户的行为序列中两个商品同时出现的次数。例如用户 U1 购买了 A 和 B则 A-B 共现次数加 1。MapReduce 的第一步需要把清洗后的数据按用户分组然后枚举该用户所有商品的两两组合。Mapper 读取清洗后的userId \t itemId \t behavior输出 key 为userIdvalue 为itemId。Reducer 端需要注意如果用户购买的商品数量很多一个用户下的配对数量会呈平方级增长可能造成 OOM。常见做法是设置job.setJar后利用setup方法中的 max 阈值截断或者只保留该用户权重最高的前 N 个商品通常在 50 以内。public class CoOccurrenceReducer extends ReducerText, Text, Text, IntWritable { private Text pairKey new Text(); private IntWritable one new IntWritable(1); Override protected void reduce(Text userId, IterableText itemIds, Context context) throws IOException, InterruptedException { ListString items new ArrayList(); // 只取前 50 个商品防止配对爆炸 int maxItems 50; for (Text item : itemIds) { if (items.size() maxItems) break; items.add(item.toString()); } // 生成所有两两组合i j 的情况跳过 for (int i 0; i items.size(); i) { for (int j i 1; j items.size(); j) { String itemI items.get(i); String itemJ items.get(j); if (itemI.equals(itemJ)) continue; // 输出 pairKey 时保证字典序避免 A-B 和 B-A 重复统计 if (itemI.compareTo(itemJ) 0) { pairKey.set(itemI \t itemJ); } else { pairKey.set(itemJ \t itemI); } context.write(pairKey, one); } } } }这里交换itemI和itemJ顺序是个很重要的细节如果不做字典序排序A-B 和 B-A 会被当成两个 key最终共现次数变成两倍。课程设计里如果发现相似度矩阵数值异常偏大先查这里。4.3 第二阶段计算余弦相似度获得共现矩阵后需要计算相似度。经典的 Item-based CF 使用余弦相似度公式如下[ sim(i,j) \frac{|N(i) \cap N(j)|}{\sqrt{|N(i)| \cdot |N(j)|}} ]其中 (N(i)) 表示喜欢/购买了商品 i 的用户集合。分母需要在上一阶段统计每个商品出现的总次数。所以在第二阶段的 Mapper 中既要读取共现矩阵key 为itemI itemJvalue 为 count又要读取商品频次表key 为itemIdvalue 为 totalCount。一种更简洁的做法是把两个输入放到同一个作业中通过MultipleInputs为不同路径设置不同的 Mapper 类。但为了课程设计理解简单建议分两个作业第一个作业统计商品频次Map 输出itemId 1Reduce 求和。第二个作业读取共现矩阵和频次文件Mapper 缓存频次映射到job的分布式缓存cache中。使用 DistributedCache 的典型代码片段// 在 Mapper 的 setup 中读取缓存文件 protected void setup(Context context) throws IOException, InterruptedException { MapString, Integer itemFreq new HashMap(); URI[] cacheFiles context.getCacheFiles(); if (cacheFiles ! null) { for (URI cacheFile : cacheFiles) { String filename cacheFile.toString(); Path path new Path(filename); try (BufferedReader reader new BufferedReader( new InputStreamReader(FileSystem.get(context.getConfiguration()) .open(path)))) { String line; while ((line reader.readLine()) ! null) { String[] parts line.split(\t); itemFreq.put(parts[0], Integer.parseInt(parts[1])); } } } } this.itemFreq itemFreq; }然后在 reduce 阶段同时拿到count和itemFreq.get(itemI)、itemFreq.get(itemJ)套公式计算。这里要注意分母不能为 0否则会产生无穷大相似度实际工程中会加平滑因子比如分母加 1。4.4 生成 Top-N 推荐列表相似度矩阵计算完后最后一个 MapReduce 作业为每个用户生成推荐。输入是清洗后的用户行为数据 商品相似度矩阵。Mapper 把相似度矩阵按itemI为 keyvalue 为itemJ:sim输出ReadJob 中再根据用户已购买的物品来匹配候选物品。Reducer 需要过滤掉用户已经买过的商品按推荐得分排序取前 N 个。由于 Hadoop MapReduce 的 Reducer 对 value 是迭代器形式不能直接排序常见做法是在 Reducer 中用一个 TreeMap把评分作为 key 保持有序private TreeMapDouble, String topNMap new TreeMap(Collections.reverseOrder()); Override protected void reduce(Text userId, IterableText candidateScores, Context context) throws IOException, InterruptedException { topNMap.clear(); SetString userHistory new HashSet(); // 由 generateCandidates 阶段传入 for (Text score : candidateScores) { String[] s score.toString().split(:); String item s[0]; double sim Double.parseDouble(s[1]); if (userHistory.contains(item)) continue; topNMap.put(sim, item); if (topNMap.size() 10) { // 只保留 Top-10 topNMap.remove(topNMap.lastKey()); } } for (Map.EntryDouble, String entry : topNMap.entrySet()) { context.write(new Text(userId.toString() \t entry.getValue() \t entry.getKey()), NullWritable.get()); } }TreeMap 按 double 逆序排列当插入数量超过 10 时就移除队尾最小的保证当前内存中始终只有最高的 10 个。这个模式的优点是 Reducer 不用维护全量列表适合推荐列表长度固定为 N 的场景。4.5 混合推荐的实现思路如果课程设计提到混合推荐最简单的做法是在相似度计算结果之上叠加热度惩罚或商品属性相似度。比如使用加权公式final_score α * item_cf_score (1 - α) * content_score其中content_score可以通过商品分类的 one-hot 编码求向量余弦。这一步可以不用 Hadoop 计算而是把相似度矩阵导出后用 SQL 或 Python 处理。我一般会保留一个配置项CF_ALPHA在hadoop jar命令里用-Dcf.alpha0.8传入这样调参不需要重新编译代码。下面是读取参数的示例Configuration conf context.getConfiguration(); double alpha conf.getDouble(cf.alpha, 0.8);这样设计的好处是评估实验时可以快速跑多组数值而不必反复打包上传 jar 包。5. 集群部署调试与性能优化从 3 分钟到 40 秒的调优记录最后一章分享几个这个课程设计中真正影响运行效果的细节尤其是在伪分布式环境里最容易踩的坑。5.1 输入数据量过小导致的「假死」课程设计给的数据如果只有几千行往往会发生 Mapper 启动后瞬间完成、Reducer 却等待很久的现象。这是因为默认的mapreduce.reduce.memory.mb和mapreduce.reduce.cpu.vcores设置过高而任务调度器在等待资源。建议在 yarn-site.xml 中调低约束property nameyarn.nodemanager.resource.memory-mb/name value4096/value /property property nameyarn.scheduler.minimum-allocation-mb/name value256/value /property用hadoop jar提交时也可以覆盖参数hadoop jar recommend.jar CoOccurrenceJob \ -D mapreduce.reduce.memory.mb512 \ -D mapreduce.map.memory.mb512 \ -D mapreduce.job.reduces2 \ /recommend/clean /recommend/cooccurrence-D必须放在 jar 路径之前这是 Hadoop CLI 解析参数的一个位置约束很多人弄反后参数不生效还找不到问题。5.2 Combiner 的恰当使用在共现矩阵生成阶段可以在 Mapper 后添加 Combiner在 Map 端做一次局部求和job.setCombinerClass(CoOccurrenceReducer.class);但要注意Combiner 的输入输出类型必须与 Reducer 一致。如果 Reducer 输出Text, IntWritableCombiner 也必须是同一对类型。这个优化在数据量达到百万级时效果明显因为 Shuffle 传输的数据量大幅减少。但课程设计演示时最好先不加 Combiner观察 Shuffle 字节数再加 Combiner 对比这样能直观感受到优化效果。5.3 验证推荐结果的方法跑完最后一个作业后输出目录下可能有多个part-r-xxxxx文件直接查看比较费力。可以用下面命令合并指定大小的输出hdfs dfs -getmerge /recommend/output/ result.csv wc -l result.csv awk -F \t {print $2} result.csv | sort | uniq -c | sort -rn | head这可以快速确认推荐列表中是否存在高频商品异常爆炸的情况。比如某个商品出现在 90% 的用户推荐列表中说明相似度计算可能退化成热门排序需要检查权重分配或是否在生成候选时过滤掉了已购买的物品。5.4 评估指标计算脚本课程设计报告需要评估数据推荐用一个简单的 Python 脚本计算覆盖率与召回率。假设测试集的真实购买行为是user_item.csv推荐结果是result.csv# evaluate.py 用于计算推荐结果覆盖率 import sys result_path sys.argv[1] rec_items set() user_recs {} with open(result_path, r) as f: for line in f: parts line.strip().split(\t) if len(parts) 2: continue user_id, item_id parts[0], parts[1] rec_items.add(item_id) if user_id not in user_recs: user_recs[user_id] [] user_recs[user_id].append(item_id) # 覆盖率推荐出的商品数占总商品数的比例 total_items 1000 # 根据你的商品表大小修改 coverage len(rec_items) / total_items print(Coverage: {:.4f}.format(coverage))这个脚本本身不依赖 Hadoop放在项目script/目录下即可。覆盖率指标能快速判断推荐结果是否过于集中。如果覆盖率低于 20%优先检查是不是有商品从未被任何用户点击导致相似度矩阵稀疏。可以在共现矩阵结果中统计非零键数量来辅助判断。本文还有配套的精品资源点击获取
Hadoop实战：基于MapReduce的商品推荐系统设计与实现
简介针对大数据与推荐系统课程设计场景这份基于Hadoop的商品推荐系统项目包提供了完整的可参考实现适合高校学生、开发者快速上手分布式推荐系统搭建。包内共19个文件以Java源码和XML配置为主另含properties环境配置、README说明及txt文档合计26KB结构简洁便于直接导入工程进行二次开发。课程设计内容涵盖Hadoop核心组件HDFS与MapReduce的数据处理流程以及协同过滤、基于内容、混合推荐等主流算法项目源码按数据输入层、数据处理层、模型训练层和推荐服务层分模块组织清晰呈现从用户行为数据清洗到个性化推荐结果生成的完整链路。同时给出YARN资源调度、矩阵分解等优化思路与准确率、召回率等评估指标便于学生理解推荐系统工程的落地要点。已有3435人学习下载适合作为课程报告、毕业设计或企业入门实践的参考资料。1. 为什么课程设计选它Hadoop 与推荐系统天然互补拿到这份《基于 Hadoop 商品推荐系统课程设计》压缩包时我第一反应不是看代码而是先看工程结构——GoodRecommendationManagementSystem-master这种命名方式基本能判断是教学用的 Maven 项目。真正把它跑起来后会发现这套设计把 HDFS 存储、MapReduce 计算和协同过滤推荐串在了一条完整链路上而不是像很多 Demo 那样只写个单机推荐算法交差。课程的落点很明确让你理解推荐系统在电商场景下的数据流动同时把 Hadoop 分布式能力真的用起来。适合正在做大数据方向课程设计、或者想把 Hadoop 生态应用在具体业务上的同学也适合想回看经典 MapReduce 编程模型的工程师。从解压到跑通每一步都需要动手改配置、调参数这也是它的价值所在。2. 系统架构从用户点击到推荐结果的四层数据链路拿到项目后建议先不急着看 Mapper 和 Reducer而是对照README.md理清系统的分层。一个能跑通的 Hadoop 推荐系统必然不是把算法代码扔到集群上就行它需要对数据输入、存储、计算和输出做完整设计。2.1 四层架构与 Hadoop 组件对应关系这个项目的核心架构可以拆成四层每一层都能在 Hadoop 生态里找到对应组件下面是实际工程中最常见的映射关系层级职责Hadoop 组件典型数据形态数据输入层收集用户行为日志、商品信息写入分布式存储HDFS Client / Flume用户 ID、商品 ID、行为类型、时间戳数据处理层清洗、过滤、聚合原始数据生成推荐算法所需的中间表MapReduce / Hive去重后的行为记录、用户-商品评分矩阵模型训练层计算商品相似度、用户偏好向量生成推荐模型MapReduce 多轮作业商品共现矩阵、相似度 Top-K 列表推荐服务层接收查询请求根据模型匹配 Top-N 结果返回给前端HBase / Redis / 纯 Java API用户 ID 对应的推荐列表很多初学者会忽略一个关键点推荐服务层其实不在 Hadoop 的核心计算链路里它更多是读取模型结果。所以在这个课程设计中模型训练的产出物——比如商品相似度文件或用户推荐列表——需要落回 HDFS 或者导出到本地文件才能被上层应用使用。2.2 HDFS 目录规划与数据预处理存储项目的输入数据通常是一份用户行为日志字段格式类似userId,itemId,behaviorType,timestamp。这里有一个常见坑如果直接把原始日志丢给推荐算法结果会非常差因为日志中有大量浏览行为、重复点击、以及没有登录的游客 ID。所以必须先做数据清洗。我一般会在 HDFS 上建三个目录分别存放原始数据、清洗后数据、推荐模型输出# 创建 HDFS 目录依次存放原始日志、清洗结果、模型输出 hdfs dfs -mkdir -p /recommend/input hdfs dfs -mkdir -p /recommend/clean hdfs dfs -mkdir -p /recommend/output # 上传本地日志文件到 HDFS hdfs dfs -put user_behavior.log /recommend/input/这里说下为什么划分这三个目录input目录只做数据落盘不参与计算clean目录存储 MapReduce 清洗作业的输出供后续相似度计算作业读取output目录存放每轮作业的结果。如果所有作业的输入输出都指向同一个目录MapReduce 的OutputCommitter会直接报错这也是 Hadoop 编程中非常容易踩的坑。2.3 数据表结构与字段约束在写 MapReduce 之前你需要明确输入数据的字段顺序和类型。这个项目的数据格式虽然简单但强烈建议你在 README 里补一张字段约束表否则换一套数据就很容易出现ClassCastException。字段名类型示例约束说明user_idStringU12345不能为空游客 ID 需过滤item_idStringI789商品 ID需要与商品表关联behavior_typeStringclick / buy / collect不同行为对应不同权重timestampLong1612137600Unix 时间戳用于时间窗口过滤这里有一个设计决策为什么不直接使用原始的浏览记录我个人的做法是在清洗阶段做行为去重同一个用户在同一天对同一个商品的多次点击只保留一次但buy行为不合并。原因是buy行为的置信度远高于click去重会损失关键特征。3. 数据清洗与用户-商品矩阵构建写第一个生产级 MapReduce 作业这一章是整个项目中的基础也是 Hadoop 入门者对 MapReduce 理解最容易出偏差的地方。课程设计的核心不是调包调参而是能自己写一个从原始日志到用户-商品行为矩阵的完整作业。3.1 清洗逻辑设计什么数据该过滤在写 Mapper 之前先定义清洗规则。我建议至少做这四件事:过滤掉user_id为空的记录过滤掉behavior_type不在预设集合click、buy、collect中的记录对同一用户对同一商品的重复点击记录去重过滤掉时间戳超出业务范围的过期数据这些规则用 MapReduce 实现时map阶段负责解析和过滤reduce阶段负责按userId itemId做去重聚合。这样的设计能让 Map 端通过context.write输出的中间数据量显著减少减轻 Shuffle 的压力。3.2 Mapper 实现与参数说明下面是清洗作业的 Java Mapper 代码这个写法在 Hadoop 2/3 中都可以直接编译运行:import java.io.IOException; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; public class CleanMapper extends MapperLongWritable, Text, Text, Text { // 输出 key 使用 userId_itemIdvalue 使用行为类型加时间戳 private Text outKey new Text(); private Text outValue new Text(); Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line value.toString().trim(); if (line.isEmpty()) { return; } String[] fields line.split(,); // 字段数量不足或 user_id 为空直接丢弃 if (fields.length 4 || fields[0].isEmpty()) { return; } String userId fields[0]; String itemId fields[1]; String behavior fields[2]; String timestamp fields[3]; // 只保留已知的行为类型避免脏数据参与计算 if (!click.equals(behavior) !buy.equals(behavior) !collect.equals(behavior)) { return; } outKey.set(userId _ itemId); // 将行为和原始时间戳拼接作为 value outValue.set(behavior \t timestamp); context.write(outKey, outValue); } }这里关键参数是 Mapper 的泛型LongWritable, Text, Text, Text输入是行偏移量和文本输出 key 是用户商品组合键value 是行为与时间戳。为什么不用TextPair做组合键因为这里不需要在 Map 端排序后按用户分组直接用字符串拼接减少自定义 WritableComparable 的复杂度适合课程设计阶段的体量。3.3 Reducer 去重与生成行为向量Reducer 端的工作是处理同一个userId_itemId下的多条记录。由于 Map 输出已经按 key 分组Reducer 只需遍历 values保留权重最高的行为。比如buy的优先级大于collect大于click。import java.io.IOException; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; public class CleanReducer extends ReducerText, Text, Text, NullWritable { Override protected void reduce(Text key, IterableText values, Context context) throws IOException, InterruptedException { String bestBehavior click; // 默认最低优先级 long bestTimestamp 0L; for (Text val : values) { String[] parts val.toString().split(\t); if (parts.length ! 2) { continue; } String behavior parts[0]; long ts Long.parseLong(parts[1]); // 行为优先级判断buy collect click int currentPriority getPriority(behavior); if (currentPriority getPriority(bestBehavior) || (currentPriority getPriority(bestBehavior) ts bestTimestamp)) { bestBehavior behavior; bestTimestamp ts; } } // 输出格式userId_itemId behavior String[] keyParts key.toString().split(_); context.write(new Text(keyParts[0] \t keyParts[1] \t bestBehavior), NullWritable.get()); } private int getPriority(String behavior) { if (buy.equals(behavior)) return 3; if (collect.equals(behavior)) return 2; return 1; } }这里有两个值得注意的细节。第一Reducer 的 key 是Text类型但实际 value 中包含了下划线分隔的 userId 和 itemId所以输出时做了split(_)。第二NullWritable作为输出 value 表示我们不需要保留中间聚合值直接写最终文本。如果你后续要读入 Hive 表建议改成Text输出并加一个列分隔符比如\t。3.4 提交作业与观察运行状态写完 Maven 项目后需要把代码打成 jar 包并提交到集群。课程设计环境中通常是伪分布式直接用hadoop jar命令提交即可# 打包跳过测试避免 test 类干扰 mvn clean package -DskipTests # 提交清洗作业指定输入输出目录 hadoop jar target/recommend-system-1.0.jar CleanJob /recommend/input /recommend/clean # 查看输出结果 hdfs dfs -cat /recommend/clean/part-r-00000 | head -20命令中的CleanJob是包含main方法的类名需要在其中设置setMapperClass、setReducerClass、setOutputKeyClass等参数。这里我补充一个容易被忽略的参数job.setNumReduceTasks(2)。伪分布式默认只有一个 Reducer但如果你希望多个 Reduce 任务以加深对 Shuffle 的理解可以显式设置为 2。但要注意Reduce 任务数不是越多越好每个 Reducer 会分别输出一个part-r-xxxxx文件下游作业需要通配读取。4. 基于物品的协同过滤用 MapReduce 分步实现相似度计算与 Top-N 推荐课程设计的核心算法是协同过滤。很多人一上来就写一个cosineSimilarity的类但放到 Hadoop 上就必须拆解成多个 MapReduce 作业——因为相似度矩阵的规模可能远超单机内存。4.1 算法选型为什么用 Item-based CF 而不是 User-based CF在电商场景中物品数量通常远小于用户数量。以 10 万用户、1 万商品为例User-based CF 需要计算用户之间的相似度矩阵规模是 10 万 × 10 万而 Item-based CF 只需要计算 1 万 × 1 万。Hadoop 的 MapReduce 本质上更适合「先分割再合并」的计算Item-based 的每两两商品组合可以用一个 Reduce 任务处理天然符合分布式模型。因此这个课程设计选用 Item-based CF 是合理的选择。4.2 第一阶段构造商品共现矩阵共现矩阵的含义是在同一用户的行为序列中两个商品同时出现的次数。例如用户 U1 购买了 A 和 B则 A-B 共现次数加 1。MapReduce 的第一步需要把清洗后的数据按用户分组然后枚举该用户所有商品的两两组合。Mapper 读取清洗后的userId \t itemId \t behavior输出 key 为userIdvalue 为itemId。Reducer 端需要注意如果用户购买的商品数量很多一个用户下的配对数量会呈平方级增长可能造成 OOM。常见做法是设置job.setJar后利用setup方法中的 max 阈值截断或者只保留该用户权重最高的前 N 个商品通常在 50 以内。public class CoOccurrenceReducer extends ReducerText, Text, Text, IntWritable { private Text pairKey new Text(); private IntWritable one new IntWritable(1); Override protected void reduce(Text userId, IterableText itemIds, Context context) throws IOException, InterruptedException { ListString items new ArrayList(); // 只取前 50 个商品防止配对爆炸 int maxItems 50; for (Text item : itemIds) { if (items.size() maxItems) break; items.add(item.toString()); } // 生成所有两两组合i j 的情况跳过 for (int i 0; i items.size(); i) { for (int j i 1; j items.size(); j) { String itemI items.get(i); String itemJ items.get(j); if (itemI.equals(itemJ)) continue; // 输出 pairKey 时保证字典序避免 A-B 和 B-A 重复统计 if (itemI.compareTo(itemJ) 0) { pairKey.set(itemI \t itemJ); } else { pairKey.set(itemJ \t itemI); } context.write(pairKey, one); } } } }这里交换itemI和itemJ顺序是个很重要的细节如果不做字典序排序A-B 和 B-A 会被当成两个 key最终共现次数变成两倍。课程设计里如果发现相似度矩阵数值异常偏大先查这里。4.3 第二阶段计算余弦相似度获得共现矩阵后需要计算相似度。经典的 Item-based CF 使用余弦相似度公式如下[ sim(i,j) \frac{|N(i) \cap N(j)|}{\sqrt{|N(i)| \cdot |N(j)|}} ]其中 (N(i)) 表示喜欢/购买了商品 i 的用户集合。分母需要在上一阶段统计每个商品出现的总次数。所以在第二阶段的 Mapper 中既要读取共现矩阵key 为itemI itemJvalue 为 count又要读取商品频次表key 为itemIdvalue 为 totalCount。一种更简洁的做法是把两个输入放到同一个作业中通过MultipleInputs为不同路径设置不同的 Mapper 类。但为了课程设计理解简单建议分两个作业第一个作业统计商品频次Map 输出itemId 1Reduce 求和。第二个作业读取共现矩阵和频次文件Mapper 缓存频次映射到job的分布式缓存cache中。使用 DistributedCache 的典型代码片段// 在 Mapper 的 setup 中读取缓存文件 protected void setup(Context context) throws IOException, InterruptedException { MapString, Integer itemFreq new HashMap(); URI[] cacheFiles context.getCacheFiles(); if (cacheFiles ! null) { for (URI cacheFile : cacheFiles) { String filename cacheFile.toString(); Path path new Path(filename); try (BufferedReader reader new BufferedReader( new InputStreamReader(FileSystem.get(context.getConfiguration()) .open(path)))) { String line; while ((line reader.readLine()) ! null) { String[] parts line.split(\t); itemFreq.put(parts[0], Integer.parseInt(parts[1])); } } } } this.itemFreq itemFreq; }然后在 reduce 阶段同时拿到count和itemFreq.get(itemI)、itemFreq.get(itemJ)套公式计算。这里要注意分母不能为 0否则会产生无穷大相似度实际工程中会加平滑因子比如分母加 1。4.4 生成 Top-N 推荐列表相似度矩阵计算完后最后一个 MapReduce 作业为每个用户生成推荐。输入是清洗后的用户行为数据 商品相似度矩阵。Mapper 把相似度矩阵按itemI为 keyvalue 为itemJ:sim输出ReadJob 中再根据用户已购买的物品来匹配候选物品。Reducer 需要过滤掉用户已经买过的商品按推荐得分排序取前 N 个。由于 Hadoop MapReduce 的 Reducer 对 value 是迭代器形式不能直接排序常见做法是在 Reducer 中用一个 TreeMap把评分作为 key 保持有序private TreeMapDouble, String topNMap new TreeMap(Collections.reverseOrder()); Override protected void reduce(Text userId, IterableText candidateScores, Context context) throws IOException, InterruptedException { topNMap.clear(); SetString userHistory new HashSet(); // 由 generateCandidates 阶段传入 for (Text score : candidateScores) { String[] s score.toString().split(:); String item s[0]; double sim Double.parseDouble(s[1]); if (userHistory.contains(item)) continue; topNMap.put(sim, item); if (topNMap.size() 10) { // 只保留 Top-10 topNMap.remove(topNMap.lastKey()); } } for (Map.EntryDouble, String entry : topNMap.entrySet()) { context.write(new Text(userId.toString() \t entry.getValue() \t entry.getKey()), NullWritable.get()); } }TreeMap 按 double 逆序排列当插入数量超过 10 时就移除队尾最小的保证当前内存中始终只有最高的 10 个。这个模式的优点是 Reducer 不用维护全量列表适合推荐列表长度固定为 N 的场景。4.5 混合推荐的实现思路如果课程设计提到混合推荐最简单的做法是在相似度计算结果之上叠加热度惩罚或商品属性相似度。比如使用加权公式final_score α * item_cf_score (1 - α) * content_score其中content_score可以通过商品分类的 one-hot 编码求向量余弦。这一步可以不用 Hadoop 计算而是把相似度矩阵导出后用 SQL 或 Python 处理。我一般会保留一个配置项CF_ALPHA在hadoop jar命令里用-Dcf.alpha0.8传入这样调参不需要重新编译代码。下面是读取参数的示例Configuration conf context.getConfiguration(); double alpha conf.getDouble(cf.alpha, 0.8);这样设计的好处是评估实验时可以快速跑多组数值而不必反复打包上传 jar 包。5. 集群部署调试与性能优化从 3 分钟到 40 秒的调优记录最后一章分享几个这个课程设计中真正影响运行效果的细节尤其是在伪分布式环境里最容易踩的坑。5.1 输入数据量过小导致的「假死」课程设计给的数据如果只有几千行往往会发生 Mapper 启动后瞬间完成、Reducer 却等待很久的现象。这是因为默认的mapreduce.reduce.memory.mb和mapreduce.reduce.cpu.vcores设置过高而任务调度器在等待资源。建议在 yarn-site.xml 中调低约束property nameyarn.nodemanager.resource.memory-mb/name value4096/value /property property nameyarn.scheduler.minimum-allocation-mb/name value256/value /property用hadoop jar提交时也可以覆盖参数hadoop jar recommend.jar CoOccurrenceJob \ -D mapreduce.reduce.memory.mb512 \ -D mapreduce.map.memory.mb512 \ -D mapreduce.job.reduces2 \ /recommend/clean /recommend/cooccurrence-D必须放在 jar 路径之前这是 Hadoop CLI 解析参数的一个位置约束很多人弄反后参数不生效还找不到问题。5.2 Combiner 的恰当使用在共现矩阵生成阶段可以在 Mapper 后添加 Combiner在 Map 端做一次局部求和job.setCombinerClass(CoOccurrenceReducer.class);但要注意Combiner 的输入输出类型必须与 Reducer 一致。如果 Reducer 输出Text, IntWritableCombiner 也必须是同一对类型。这个优化在数据量达到百万级时效果明显因为 Shuffle 传输的数据量大幅减少。但课程设计演示时最好先不加 Combiner观察 Shuffle 字节数再加 Combiner 对比这样能直观感受到优化效果。5.3 验证推荐结果的方法跑完最后一个作业后输出目录下可能有多个part-r-xxxxx文件直接查看比较费力。可以用下面命令合并指定大小的输出hdfs dfs -getmerge /recommend/output/ result.csv wc -l result.csv awk -F \t {print $2} result.csv | sort | uniq -c | sort -rn | head这可以快速确认推荐列表中是否存在高频商品异常爆炸的情况。比如某个商品出现在 90% 的用户推荐列表中说明相似度计算可能退化成热门排序需要检查权重分配或是否在生成候选时过滤掉了已购买的物品。5.4 评估指标计算脚本课程设计报告需要评估数据推荐用一个简单的 Python 脚本计算覆盖率与召回率。假设测试集的真实购买行为是user_item.csv推荐结果是result.csv# evaluate.py 用于计算推荐结果覆盖率 import sys result_path sys.argv[1] rec_items set() user_recs {} with open(result_path, r) as f: for line in f: parts line.strip().split(\t) if len(parts) 2: continue user_id, item_id parts[0], parts[1] rec_items.add(item_id) if user_id not in user_recs: user_recs[user_id] [] user_recs[user_id].append(item_id) # 覆盖率推荐出的商品数占总商品数的比例 total_items 1000 # 根据你的商品表大小修改 coverage len(rec_items) / total_items print(Coverage: {:.4f}.format(coverage))这个脚本本身不依赖 Hadoop放在项目script/目录下即可。覆盖率指标能快速判断推荐结果是否过于集中。如果覆盖率低于 20%优先检查是不是有商品从未被任何用户点击导致相似度矩阵稀疏。可以在共现矩阵结果中统计非零键数量来辅助判断。本文还有配套的精品资源点击获取
