先说结论这是一个非常适合练手离线计算的项目。社交推荐、个性化推荐在业务里被包装得很玄乎但落到 MapReduce 这个层面二度好友推荐的本质就一句话把“共同好友”数清楚。你只要能把用户的好友列表拆成两两组合再用 Reduce 阶段按组合做聚合整条链路就通了。今天这篇我按自己在 CentOS 7 Hadoop 2.6.0 IDEA 环境里实际跑通的流程把这个案例从头到尾拆一遍包括两个 Job 的设计思路、完整可运行的代码、运行命令以及我踩过的几个典型坑。1. 需求拆解二度好友推荐到底在算什么1.1 从“朋友的朋友”说起社交网络里的好友推荐日常最常见的一种就是“你可能认识的人”。它的计算基础很简单如果 A 和 B 都是你的好友那么 A 和 B 之间大概率也认识系统就会把 B 推荐给 A。这里的“度”怎么理解我自己的好友是一度关系我好友的好友是二度关系。二度好友推荐要做的就是找出那些和我不是直接好友、但和我共享了至少一个共同好友的人然后按共同好友的数量从多到少排序优先推荐给我。这个逻辑用生活类比解释你参加一个聚会现场十个人里你只认识三个人这时候你会优先和谁搭话大概率是那个和你认识的人重叠最多的人。因为共同认识的人越多你们之间有共同话题的概率就越高。这也是为什么二度好友推荐在社交产品里转化率一直不错。放到 MapReduce 框架下这个需求很适合用来练手。数据输入是一堆“用户 好友列表”的文本行处理过程不涉及复杂的图遍历核心就是分组聚合。它天然能拆成两个问题第一哪些人之间存在二度关系第二每个人和候选好友之间有多少个共同好友这两个问题刚好对应两个 MapReduce Job。1.2 一条推荐记录是怎么被算出来的我用一个具体例子说明。假设输入数据长这样小明 小红 小刚 小红 小明 小丽 小刚 小明 小丽 小丽 小红 小刚第一列是用户后面是 TA 的好友列表。现在看“小明”这一行他的好友是“小红”和“小刚”那么“小红”和“小刚”这两个人之间就通过“小明”产生了一次关联。换句话说小明是这两个人的共同好友。同理小红的好友列表里有小明和小丽那么“小明”和“小丽”之间也通过小红产生了一次关联。把每一次“两两组合”都输出出去再按组合做聚合最后统计每个组合被多少个不同的用户提到过这个次数就是两个候选用户之间的共同好友数量。但这里有一个关键前提如果这两个人本来就是直接好友那就不能推荐了。比如数据里小明的好友列表里有小红小红的好友列表里也有小明说明他们已经是一度好友应该过滤掉。所以整个计算过程实际分三步把每个用户的好友列表做两两组合输出“组合 - 共同好友”把互为直接好友的关系也单独标记出来聚合时如果发现这个组合已经是直接好友直接丢弃否则统计共同好友数量作为推荐分数。这三步刚好可以用两个 Job 完成这也是我建议你在练手时采用的结构。第一个 Job 负责找关系和数共同好友第二个 Job 负责按用户分组、按共同好友数排序输出 TopN。2. 环境准备CentOS 7 Hadoop 2.6.0 IDEA 搭起来2.1 版本选型与依赖配置先说明一下版本选择。Hadoop 2.6.0 对很多人来说是“老古董”但它搭配 CentOS 7 和 JDK 1.8 是一套非常稳的组合。很多教材、企业内部保留的老集群都跑在这个版本上网上资料也足够多遇到问题容易搜到答案。相比 Hadoop 3.x2.6.0 的 API 在某些地方有差异比如部分类所在的包路径、部分参数名称但这些差异不影响我们理解 MapReduce 的核心思想。如果你上来直接用 3.x反而可能在版本兼容上多花时间。在 IDEA 里创建 Maven 工程后pom.xml 里只加一个核心依赖dependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-client/artifactId version2.6.0/version scopeprovided/scope /dependency这里 scope 用 provided意思是编译和测试时依赖它但打包时不把 Hadoop 相关类打进去。因为集群环境里本来就有 Hadoop 的 jar如果你把 Hadoop 依赖一起打成 fat jar 再往集群上提交轻则类冲突重则直接起不来。这个坑我一开始就踩过后来才明白集群运行的 jar 应该保持干净。打包插件我用的是 maven-jar-plugin直接指定主类plugin groupIdorg.apache.maven.plugins/groupId artifactIdmaven-jar-plugin/artifactId configuration archive manifest mainClasscom.example.recommend.FriendRecommend/mainClass /manifest /archive /configuration /plugin如果你要在 Windows 本机直接跑 MapReduce还需要额外配置 winutils.exe 和 HADOOP_HOME那套东西很烦。这个案例的推荐做法是先在 IDEA 里写好代码、编译通过然后打成 jar 传到 Linux 集群机器上跑。这样既不用折腾 Windows 下的本地环境也更接近真实生产流程。2.2 伪分布式集群部署关键点如果你手头没有现成的集群在 CentOS 7 上搭一个伪分布式环境就够了。所谓伪分布式就是所有进程——NameNode、DataNode、ResourceManager、NodeManager——都跑在同一台机器上对学习 MapReduce 完全够用。部署顺序大致是安装 JDK 1.8配置 JAVA_HOME下载 Hadoop 2.6.0 安装包并解压修改 etc/hadoop/hadoop-env.sh把 JAVA_HOME 写死修改 core-site.xml、hdfs-site.xml、mapred-site.xml、yarn-site.xml执行 hdfs namenode -format 格式化元数据执行 start-dfs.sh 和 start-yarn.sh 启动集群用 jps 查看进程确认 NameNode、DataNode、ResourceManager、NodeManager 都起来了。有四个配置我会重点检查。core-site.xml 里设置 fs.defaultFS 为 hdfs://localhost:9000hdfs-site.xml 里把 dfs.replication 改成 1伪分布式只有一台机器副本数填 3 会一直报副本不足mapred-site.xml 把 mapreduce.framework.name 设为 yarnyarn-site.xml 里确保 ResourceManager 的 hostname 指向本机。启动后先用 jps 看进程再访问 50070 端口看 HDFS 界面、8088 端口看 YARN 界面。如果页面打不开先检查 CentOS 7 的防火墙systemctl stop firewalld systemctl disable firewalld这个案例里我直接关了防火墙省心。如果你不想关防火墙就把 50070、8088、9000、8042 这些端口加到白名单里但说实话练手环境直接关掉最省事。3. 核心代码两个 Job 串联实现完整推荐流程3.1 Job1找出所有二度关系并统计共同好友数Job1 是整个案例的心脏。Mapper 输入一行文本key 是行偏移量value 是“用户 好友1 好友2 …”这样的字符串。我在 Mapper 里做两件事。第一件给当前用户的每个好友生成一条 DIRECT 标记表示这两个人是直接好友。这里的 key 是“好友对”value 是固定字符串 DIRECT。第二件把当前用户的好友列表做两两组合每个组合输出的 key 也是“好友对”value 是当前用户的名字。当前用户就是这对候选好友的共同好友。这里有一个容易忽略的细节好友对里的两个名字必须排序后再拼成 key。如果 A 比 B 字母序小key 永远是“A\tB”而不是有时候“A\tB”有时候“B\tA”。不排序的话同一对人在 Shuffle 阶段会被分到不同的 key 上共同好友永远统计不到一起。这个归一化处理是整个算法的根基。Reducer 做的事也不复杂。遍历某个好友对的所有 value如果里面出现了 DIRECT说明这俩人已经是直接好友直接跳过。否则把所有非 DIRECT 的 value 收集到一个 Set 里Set 的大小就是共同好友数量。数量大于 0 时把结果双向输出即 A 推荐 B、B 也推荐 A。下面是 Job1 的核心代码package com.example.recommend; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; import java.util.*; public class FriendRecommend { // 用于标记直接好友关系的特殊字符串 private static final String DIRECT_FLAG DIRECT; public static class CoFriendMapper extends MapperLongWritable, Text, Text, Text { private Text pairKey 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[] parts line.split(\\s); if (parts.length 2) { return; } String user parts[0]; SetString friends new HashSetString( Arrays.asList(Arrays.copyOfRange(parts, 1, parts.length))); // 排除好友列表里指向自己的情况 friends.remove(user); // 第一步输出直接好友关系标记 for (String friend : friends) { emitPair(user, friend, DIRECT_FLAG, context); } // 第二步好友列表两两组合输出“候选好友对 - 当前用户” ListString friendList new ArrayListString(friends); for (int i 0; i friendList.size(); i) { for (int j i 1; j friendList.size(); j) { emitPair(friendList.get(i), friendList.get(j), user, context); } } } private void emitPair(String a, String b, String value, Context context) throws IOException, InterruptedException { int cmp a.compareTo(b); if (cmp 0) { pairKey.set(a \t b); } else if (cmp 0) { pairKey.set(b \t a); } else { // a 和 b 相同跳过 return; } outValue.set(value); context.write(pairKey, outValue); } } public static class CoFriendReducer extends ReducerText, Text, Text, Text { private Text outKey new Text(); private Text outValue new Text(); Override protected void reduce(Text key, IterableText values, Context context) throws IOException, InterruptedException { boolean isDirect false; SetString commonFriends new HashSetString(); for (Text value : values) { String str value.toString(); if (DIRECT_FLAG.equals(str)) { isDirect true; } else { commonFriends.add(str); } } // 已经是直接好友不推荐 if (isDirect) { return; } int commonCount commonFriends.size(); if (commonCount 0) { return; } String[] pair key.toString().split(\t); if (pair.length ! 2) { return; } String userA pair[0]; String userB pair[1]; String countStr String.valueOf(commonCount); // 双向输出A 推荐 BB 推荐 A outKey.set(userA); outValue.set(userB \t countStr); context.write(outKey, outValue); outKey.set(userB); outValue.set(userA \t countStr); context.write(outKey, outValue); } } }这里有个小程序设计要点Reducer 输出的是 user 作为 key、候选好友加共同好友数作为 value。这样 Job2 的输入就已经按用户聚好类了省去再按用户分组的麻烦。3.2 Job2按共同好友数排序输出每个用户 TopNJob1 输出的每一行是“用户 候选好友 共同好友数”但同一个用户可能对应多个候选好友而且同一对候选好友在 Job1 里可能从不同路径输出多次所以 Job2 的 Mapper 需要把每一行拆成 user 和 candidate\countReducer 端先按 user 聚合。TopNReducer 里的聚合逻辑是把每个候选好友对应的共同好友数收集到 HashMap 中因为同一个候选好友可能出现在多个 value 里这里用 merge 求最大值。然后把这个 HashMap 转成 List 做排序排序规则是共同好友数降序数量相同时按候选好友字典序升序。最后截取前 N 个拼成一行输出。这个方案很简单但也有个前提Job2 的 Reducer 数量必须设为 1。因为 TreeMap 的排序只在单个 Reducer 内部有效如果有多个 Reducer每个 Reducer 只拿到一部分用户的数据输出的 TopN 就不是全局 TopN。案例数据量小一个 Reducer 完全够用。如果以后数据量大可以换成全排序方案比如先全局按共同好友数排序再分区取 TopN这里先不做扩展。Job2 的代码如下public static class TopNMapper extends MapperLongWritable, Text, Text, Text { private Text outKey new Text(); private Text outValue new Text(); Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] parts value.toString().split(\t); if (parts.length ! 2) { return; } outKey.set(parts[0]); outValue.set(parts[1]); context.write(outKey, outValue); } } public static class TopNReducer extends ReducerText, Text, Text, Text { private Text outValue new Text(); private int topN 10; Override protected void setup(Context context) { topN context.getConfiguration().getInt(recommend.topn, 10); } Override protected void reduce(Text key, IterableText values, Context context) throws IOException, InterruptedException { MapString, Integer candidateMap new HashMapString, Integer(); for (Text value : values) { String[] parts value.toString().split(\t); if (parts.length ! 2) { continue; } String candidate parts[0]; int count Integer.parseInt(parts[1]); // 同一个候选可能出现多条记录取最大共同好友数 Integer old candidateMap.get(candidate); if (old null || old count) { candidateMap.put(candidate, count); } } ListMap.EntryString, Integer list new ArrayListMap.EntryString, Integer(candidateMap.entrySet()); Collections.sort(list, new ComparatorMap.EntryString, Integer() { public int compare(Map.EntryString, Integer e1, Map.EntryString, Integer e2) { int cmp e2.getValue().compareTo(e1.getValue()); if (cmp ! 0) { return cmp; } return e1.getKey().compareTo(e2.getKey()); } }); StringBuilder sb new StringBuilder(); int limit Math.min(topN, list.size()); for (int i 0; i limit; i) { if (i 0) { sb.append(, ); } sb.append(list.get(i).getKey()) .append(() .append(list.get(i).getValue()) .append()); } outValue.set(sb.toString()); context.write(key, outValue); } }主函数负责串联两个 Job这里有一个必须处理的细节MapReduce 要求输出目录不能提前存在所以我在提交作业前先把临时目录和最终输出目录删除避免上一次运行留下的目录导致任务失败。public static void main(String[] args) throws Exception { if (args.length ! 3) { System.err.println(Usage: FriendRecommend input tempOutput finalOutput); System.exit(1); } Configuration conf new Configuration(); String input args[0]; String tempOutput args[1]; String finalOutput args[2]; Path tempPath new Path(tempOutput); tempPath.getFileSystem(conf).delete(tempPath, true); Path finalPath new Path(finalOutput); finalPath.getFileSystem(conf).delete(finalPath, true); // Job1统计二度好友关系和共同好友数 Job job1 Job.getInstance(conf, Friend Recommend - Find Common Friends); job1.setJarByClass(FriendRecommend.class); job1.setMapperClass(CoFriendMapper.class); job1.setReducerClass(CoFriendReducer.class); job1.setMapOutputKeyClass(Text.class); job1.setMapOutputValueClass(Text.class); job1.setOutputKeyClass(Text.class); job1.setOutputValueClass(Text.class); FileInputFormat.setInputPaths(job1, new Path(input)); FileOutputFormat.setOutputPath(job1, tempPath); if (!job1.waitForCompletion(true)) { System.exit(1); } // Job2按共同好友数排序输出每个用户的 TopN Job job2 Job.getInstance(conf, Friend Recommend - TopN); job2.setJarByClass(FriendRecommend.class); job2.setMapperClass(TopNMapper.class); job2.setReducerClass(TopNReducer.class); job2.setMapOutputKeyClass(Text.class); job2.setMapOutputValueClass(Text.class); job2.setOutputKeyClass(Text.class); job2.setOutputValueClass(Text.class); // 为了演示全局 TopN这里固定 1 个 Reducer job2.setNumReduceTasks(1); FileInputFormat.setInputPaths(job2, tempPath); FileOutputFormat.setOutputPath(job2, finalPath); System.exit(job2.waitForCompletion(true) ? 0 : 1); } }注意我在 Job2 的 Reducer setup 里读取了 recommend.topn 参数也就是说运行时可以通过-Drecommend.topn5这样的参数控制每个用户输出多少条推荐。这种把业务参数放进 Configuration 而不是写死在代码里的习惯建议从一开始就养成后面迭代需求时会省很多事。3.3 打包与集群运行代码写好后在 IDEA 里执行 mvn clean package把打出来的 jar 上传到 CentOS 7 服务器。然后准备输入数据放进 HDFShadoop fs -mkdir -p /user/hadoop/input hadoop fs -put friends.txt /user/hadoop/input/运行命令hadoop jar friend-recommend-1.0.jar \ /user/hadoop/input \ /user/hadoop/output/temp \ /user/hadoop/output/result如果你 jar 包里的 MANIFEST 没有配置 Main-Class也可以显式指定hadoop jar friend-recommend-1.0.jar com.example.recommend.FriendRecommend \ /user/hadoop/input \ /user/hadoop/output/temp \ /user/hadoop/output/result运行期间建议在浏览器里盯一下 YARN 的 8088 页面能看到两个 Job 先后执行。第一个 Job 是 CoFriendMapper 到 CoFriendReducer第二个 Job 是 TopNMapper 到 TopNReducer。Map 和 Reduce 的进度、处理记录数都能看到这也是排查问题最直观的地方。我前面那个测试数据跑完最终输出类似这样小明 小丽(2) 小红 小刚(2) 小刚 小红(2) 小丽 小明(2)含义是小明和小丽之间有 2 个共同好友所以相互推荐小红和小刚之间也有 2 个共同好友同样相互推荐。输出结果和手工推导完全一致。4. 数据倾斜、小文件与常见报错排查4.1 冷热用户导致的计算倾斜二度好友推荐这个场景里最典型的性能问题就是热点用户。假设某个明星用户有十万好友那么他的好友列表做两两组合会产生 C(100000, 2) 约 49 亿条中间记录。这会让一个 Map Task 的处理时间远超其他 Task拖慢整个 Job。这种情况在生产中很常见绝不是杞人忧天。针对这个问题的优化手段主要有三种。第一种是在 Mapper 端做局部聚合也就是用 HashMap 先缓存一部分输出在 cleanup 阶段再统一写出减少中间数据的网络传输量。第二种是加大 Shuffle 的并行度合理设置 Combiner但 Combiner 在这个场景里不太好写因为相同的 key 合并不好合并我一般选择第一种。第三种是把超大好友列表的用户单独拉出来用特殊方式处理或者设置阈值直接截断。比如好友数超过 5000 就不参与两两组合只参与 DIRECT 标记输出这样能保住大头收益又不至于让某个 Map Task 直接跑爆。我实测下来的经验是对学习案例来说一份几千条的小数据完全不用关心倾斜问题但如果未来接入真实社交数据热点用户优化一定要提前考虑否则一个用户就能拖垮整个集群。4.2 常见报错速查表报错信息原因处理方式java.io.IOException: Mkdirs failed to create /user/hadoopHDFS 目录权限不足先执行 hadoop fs -chmod -R 777 /user/hadoop或用 hdfs 用户创建目录ClassNotFoundException: com.example.recommend.FriendRecommendjar 上传错误或主类名写错检查 jar 包内容确认主类完整路径Error: JAVA_HOME is not sethadoop-env.sh 里的 JAVA_HOME 没配置编辑 hadoop-env.sh写入 export JAVA_HOME/usr/java/jdk1.8.0_xxxWARN util.NativeCodeLoader: Unable to load native-hadoop library缺少 native 本地库不影响运行可忽略想消除就配置 hadoop-native 库路径Exception in thread main org.apache.hadoop.mapred.FileAlreadyExistsException输出目录已存在程序里提前删除输出目录或换一个输出路径Reducer 输出结果为空输入数据格式不对或好友对归一化没做好检查 Mapper 输出的 key 是否按字典序排序确认 DIRECT 标记逻辑YARN 页面能打开但任务始终不执行NodeManager 内存或队列资源不足检查 yarn-site.xml 的可用内存配置调大 yarn.nodemanager.resource.memory-mb这里把最容易犯的“输出目录已存在”重点说一下。MapReduce 的设计很规矩它不会自动覆盖已有输出目录你第二次跑同一个路径时大概率直接报错。我在 main 里已经加了 delete 逻辑但你如果直接在命令行里测试别的 Jar或者改了自己的主类一定记得自己处理这个逻辑。4.3 我的实操心得最后分享两个我在写这个案例时印象最深的点都是能直接帮你少走弯路的。第一个是关于 friend pair 的排序归一化。刚开始我自己写的时候没做排序导致 A 和 B 在一条数据里输出成 “A B”在另一条数据里输出成 “B A”。Shuffle 阶段把它们当成两个不同的 key共同好友数量永远统计不正确。排查了很久才发现问题根本不在聚合代码而在数据进入 Reducer 之前就已经被拆散了。所以在 Mapper 里生成复合 key 时一定要先对 key 的组成部分做排序这算是一个通用经验凡是遇到“两个元素共同出现”的统计类任务都能用上。第二个是 Text 对象的复用问题。Hadoop 的 Mapper 和 Reducer 实例是被复用的如果你在循环里反复使用同一个 Text 对象后面写入的值会覆盖前面写入的值。很多人第一次写 MapReduce 都遇到过所有输出值全部变成最后一个值的情况。解决办法很简单每次写入时 new 一个 Text或者像我在代码里那样在方法开头创建对象并重新 set 内容。还有一个细节Reducer 里遍历 values 时直接对 value.toString() 的结果做拷贝不要在整个循环结束后再统一处理因为 Iterable 里的 value 也可能是复用的。这一点我在带别人跑案例时强调过很多次每次都能避免一批奇怪的结果。个人建议这个案例跑通之后不要急着删代码。试着改几个参数比如把 TopN 从 10 改成 3看看输出变化或者把 Job2 的 Reducer 数量从 1 改成 3你会直观地看到每个 Reducer 只输出一部分用户的结果从而理解“全局排序 vs 局部排序”的区别。再往后可以试着把 recommendation.topn 参数做成通过命令行动态传入然后观察 YARN 页面上的资源消耗变化。这样一步步折腾下来你对 MapReduce 的理解会比看十遍文档都扎实。
空间实现)
