先说结论：这是一个非常适合练手离线计算的项目。社交推荐、个性化推荐在业务里被包装得很玄乎，但落到 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 的好友列表。现在看“小明”这一行，他的好友是“小红”和“小刚”，那么“小红”和“小刚”这两个人之间，就通过“小明”产生了一次关联。换句话说，小明是这两个人的共同好友。同理，小红的好友列表里有小明和小丽，那么“小明”和“小丽”之间也通过小红产生了一次关联。

把每一次“两两组合”都输出出去，再按组合做聚合，最后统计每个组合被多少个不同的用户提到过，这个次数就是两个候选用户之间的共同好友数量。但这里有一个关键前提：如果这两个人本来就是直接好友，那就不能推荐了。比如数据里小明的好友列表里有小红，小红的好友列表里也有小明，说明他们已经是一度好友，应该过滤掉。

所以整个计算过程实际分三步：

1. 把每个用户的好友列表做两两组合，输出“组合 -> 共同好友”；
2. 把互为直接好友的关系也单独标记出来；
3. 聚合时如果发现这个组合已经是直接好友，直接丢弃；否则统计共同好友数量，作为推荐分数。

这三步刚好可以用两个 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.x，2.6.0 的 API 在某些地方有差异，比如部分类所在的包路径、部分参数名称，但这些差异不影响我们理解 MapReduce 的核心思想。如果你上来直接用 3.x，反而可能在版本兼容上多花时间。

在 IDEA 里创建 Maven 工程后，pom.xml 里只加一个核心依赖：

<dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>2.6.0</version> <scope>provided</scope> </dependency>

这里 scope 用 provided，意思是编译和测试时依赖它，但打包时不把 Hadoop 相关类打进去。因为集群环境里本来就有 Hadoop 的 jar，如果你把 Hadoop 依赖一起打成 fat jar 再往集群上提交，轻则类冲突，重则直接起不来。这个坑我一开始就踩过，后来才明白集群运行的 jar 应该保持干净。

打包插件我用的是 maven-jar-plugin，直接指定主类：

<plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-jar-plugin</artifactId> <configuration> <archive> <manifest> <mainClass>com.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 完全够用。

部署顺序大致是：

1. 安装 JDK 1.8，配置 JAVA_HOME；
2. 下载 Hadoop 2.6.0 安装包并解压；
3. 修改 etc/hadoop/hadoop-env.sh，把 JAVA_HOME 写死；
4. 修改 core-site.xml、hdfs-site.xml、mapred-site.xml、yarn-site.xml；
5. 执行 hdfs namenode -format 格式化元数据；
6. 执行 start-dfs.sh 和 start-yarn.sh 启动集群；
7. 用 jps 查看进程，确认 NameNode、DataNode、ResourceManager、NodeManager 都起来了。

有四个配置我会重点检查。core-site.xml 里设置 fs.defaultFS 为 hdfs://localhost:9000；hdfs-site.xml 里把 dfs.replication 改成 1，伪分布式只有一台机器，副本数填 3 会一直报副本不足；mapred-site.xml 把 mapreduce.framework.name 设为 yarn；yarn-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 Mapper<LongWritable, 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]; Set<String> friends = new HashSet<String>( Arrays.asList(Arrays.copyOfRange(parts, 1, parts.length))); // 排除好友列表里指向自己的情况 friends.remove(user); // 第一步：输出直接好友关系标记 for (String friend : friends) { emitPair(user, friend, DIRECT_FLAG, context); } // 第二步：好友列表两两组合，输出“候选好友对 -> 当前用户” List<String> friendList = new ArrayList<String>(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 Reducer<Text, Text, Text, Text> { private Text outKey = new Text(); private Text outValue = new Text(); @Override protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { boolean isDirect = false; Set<String> commonFriends = new HashSet<String>(); 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 推荐 B，B 推荐 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：按共同好友数排序输出每个用户 TopN

Job1 输出的每一行是“用户 候选好友 共同好友数”，但同一个用户可能对应多个候选好友，而且同一对候选好友在 Job1 里可能从不同路径输出多次，所以 Job2 的 Mapper 需要把每一行拆成 user 和 candidate\count，Reducer 端先按 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 Mapper<LongWritable, 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 Reducer<Text, 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, Iterable<Text> values, Context context) throws IOException, InterruptedException { Map<String, Integer> candidateMap = new HashMap<String, 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); } } List<Map.Entry<String, Integer>> list = new ArrayList<Map.Entry<String, Integer>>(candidateMap.entrySet()); Collections.sort(list, new Comparator<Map.Entry<String, Integer>>() { public int compare(Map.Entry<String, Integer> e1, Map.Entry<String, 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.topn=5这样的参数控制每个用户输出多少条推荐。这种把业务参数放进 Configuration 而不是写死在代码里的习惯，建议从一开始就养成，后面迭代需求时会省很多事。

3.3 打包与集群运行

代码写好后，在 IDEA 里执行 mvn clean package，把打出来的 jar 上传到 CentOS 7 服务器。然后准备输入数据，放进 HDFS：

hadoop 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 常见报错速查表

这里把最容易犯的“输出目录已存在”重点说一下。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 的理解会比看十遍文档都扎实。