企业官网建设流程全解析

最近在做社交关系数据的离线统计分析，接到一个特别朴素的需求：在千万级用户的好友关系列表里，找出任意两个用户之间的共同好友，输出成一张两两关系表。产品那边只给了这么一句话，但真正动手之后才发现，从需求到MapReduce作业之间居然隔了这么多道坎：数据怎么清洗、键怎么设计、几个作业串联、中间结果要不要落盘、倾斜怎么防。这篇文章就把整个找共同好友的Case完整拆开，把我在过程中踩过的坑和调整过的思路写清楚，给后面要做社交关系计算的同学一个可以直接参考的底稿。

1. 需求理解与数据形态：共同好友为什么要用分布式算

1.1 原始数据长什么样

在聊方案之前，先明确数据长什么样，因为后续所有MapReduce逻辑都是建立在这个格式约定之上的。我的场景里，上游数据以一行一个用户的方式存储关注关系，格式是：

userId: friendId1,friendId2,friendId3...

举个实际一点的例子：

user001: user003,user007,user012 user002: user003,user005,user012 user003: user001,user002,user007 user007: user001,user003,user012

每一行的冒号左边是用户本身，右边是该用户关注或者说认定的好友列表。这里有一个容易忽略的点：关系是有方向的。user001出现在user003的好友列表中，不保证user003一定出现在user001的列表里。有的存量数据是双向同步的，有的只是单向关注动作。如果业务上要求严格按“好友”来算，那还得先做一次双向过滤；我的案例里产品接受关注关系作为好友关系的数据基础，因此后续就按关注列表处理，不再单独做双向校验。

还有一个很常见的脏数据问题：一个用户可能重复出现在自己的好友列表里，或者某一行结尾带了空格、空串，又或者不同记录里同一个用户的id大小写不一致。这些在单机测试时不容易暴露，一旦扔到分布式环境里，会因为字符串比较、分组键不一致而产生大量“看起来对不上”的结果。我在第一版跑批时就被这事坑过一次，所以后来统一约定：上游先输出脱敏后的userId，所有id统一小写，行内不要尾随空格，好友列表用逗号分隔且不允许空元素。

1.2 直接两层循环为什么算不动

拿到这种数据，大多数人的第一反应是：每个用户和每个用户做一次集合交集。也就是遍历所有用户对，求两边好友列表的公共部分。这个思路本身没错，错在规模上。假设一千万个用户，最坏情况是10的12次方量级的配对，即使每个人只和几百人产生边，实际存在的用户对也可能到数十亿甚至更多。用单机跑这种量级的双层循环，要么内存扛不住，要么跑上几个星期都出不了结果。

这正好是MapReduce适合的场景：把配对操作打散到分布式环境里，多个节点各自算一部分，最后再把结果合并起来。但分布式不是银弹，难点在于怎么把原始数据“翻”成Map端可以直接处理的形式。直接让每个Mapper读取两行用户数据再做交集，Reducer端需要做一次笛卡尔积式的数据拉取，网络IO和内存都会爆掉。所以正确做法是设计一个中间转换，让原本分散在数据两端的共同好友关系，因为同一个键而聚合到同一个Reducer里。

1.3 共同好友计算的本质是“倒排”

仔细分析这个需求会发现，它本质上是在求一个集合的交集：用户A的好友集合和用户B的好友集合，交集就是共同好友。单机里求两个集合的交集很简单，可放到全量数据里，A和B的组合实在太多了。

MapReduce里处理这类“两两关系”问题有一个很经典的思路：先把“谁关注了谁”转换成“谁被谁关注”，也就是倒排索引。倒排之后，同一个“被关注者”下面的所有关注者，天然就是围绕这个被关注者的一个兴趣小组。任取这个小组里的两个人，那个共同的“被关注者”就是他们的一个共同好友。这样一来，一个本来需要两两用户配对的难题，被拆成了两个可以用键值对描述的阶段，每个阶段都能被Map和Reduce天然并行化。

2. 两阶段MapReduce设计：先建倒排索引，再两两组合关注者列表

2.1 第一阶段作业：把骚扰关系整理成“被关注者到关注者列表”

第一阶段的目标是产出一份倒排索引。对输入文件里每一行“user: friend1,friend2,...”，Mapper遍历好友列表里的每个friend，输出键值对(friend, user)。比如user001那行，会输出：

(user003, user001) (user007, user001) (user012, user001)

这里的含义是：user003被user001关注了，或者说user001把user003当成了好友。Reducer端把所有拥有相同key的值收集到一起，就能得到形如：

user003: user001,user002,user007

的记录。等整个第一阶段跑完，HDFS上保存的就是一份“每个被关注者都带着自己全量关注者名单”的倒排表。

可能有人会问，这不是只做了一次数据重排吗，和共同好友有什么关系？关键就在下一阶段。一旦某个被关注者下面挂着多个关注者，那么这些关注者两两之间必然至少拥有一个共同好友，就是那个被关注者本身。倒排表一出来，等于把藏在原始数据里的“隐性共同关系”显式化了。

2.2 第二阶段作业：把关注者列表两两组合成用户对

第二阶段读取第一阶段产生的倒排索引。对每一个键值对friend -> user1,user2,user3,...，Mapper把右侧的关注者列表两两组合，输出(排序后的用户对, friend)。什么意思呢？比如倒排表里有这么一条：

user003: user001,user002,user007

那就组合出三对，user001和user002、user001和user007、user002和user007，每一对的价值都填user003。这个值的含义是“user003是这两个用户的共同好友”。

这样到了Reducer端，同一个用户对下会收集来自不同倒排键的共同好友值。比如：

user001:user002 -> user003 user001:user007 -> user003,user012

最终输出就是这样一张两两关系表，每一行是一个用户对以及他们之间所有的共同好友。产品要拿这张表做接口、做推荐候选、做运营分析，都非常直接。

这里有个特别容易踩的细节：组合用户对时必须先排序再拼串。比如user001和user002，统一写成“user001:user002”，而不是一次写“user001:user002”另一次又写“user002:user001”。否则同一个用户对被分到不同的键上，Reducer永远聚不齐，结果会翻倍且错乱。

2.3 为什么不用“好友列表两两交叉”一步算完

很多讲共同好友的教程会提供一个更短的做法：对每个用户的好友列表，直接两两生成好友对，以该用户为共同好友值。例如user001的好友列表是user003、user007、user012，那就在Mapper里输出(user003:user007, user001)、(user003:user012, user001)、(user007:user012, user001)。Reducer聚合后也能得到用户对和共同好友的关系。

这个思路少了一个作业，看起来更简洁，但有一个隐含前提：输入数据必须是双向好友关系。如果user001的好友列表里有user003，但user003的列表里查不到user001，那从user001行生成的(user003:user007, user001)就不能直接说明user003和user007都把user001当好友。换句话说，单向关系会导致结果失真。

我这次选择采用倒排索引加两阶段的方式，一方面是因为产品给的数据天然是单向关注流，另一方面是因为这种结构后续扩展起来更方便。比如要做“谁和谁共同关注了同一个话题”或者“哪些用户有重叠的购买行为”，只要把输入格式换成相应的事件数据，两阶段逻辑可以原样复用。对一份可能要反复迭代的数据分析任务来说，结构上的通用性比少跑一个作业更重要。

3. 完整代码实现：Mapper、Reducer与Driver逐段解析

3.1 第一阶段Mapper与Reducer

第一阶段的Mapper逻辑非常直接：读一行，按冒号切分，遍历右侧好友列表，输出(friend, user)。但在处理脏数据时，我比第一次写的时候多做了几件事：去掉好友为空的项、跳过和自己相同的id、用Set去重。代码如下：

import java.io.IOException; import java.util.HashSet; import java.util.Set; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; public class FriendIndexMapper extends Mapper<LongWritable, Text, Text, Text> { @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); int colon = line.indexOf(':'); if (colon < 0) { return; } String user = line.substring(0, colon).trim(); if (user.isEmpty()) { return; } String[] friends = line.substring(colon + 1).split(","); Set<String> dedup = new HashSet<>(); for (String f : friends) { String friend = f.trim(); if (friend.isEmpty() || friend.equals(user)) { continue; } dedup.add(friend); } for (String friend : dedup) { context.write(new Text(friend), new Text(user)); } } }

最开始我并没有单独维护一个Set，而是直接把每个friend都写出去。后来发现上游数据里存在大量重复好友，比如一行里user003出现了五六次，导致下游值列表严重膨胀。加一个Set去重虽然会占用一点Map端内存，但能显著减少shuffle数据量，整体收益非常明显。

第一阶段Reducer要做的事情是把同一个key下的用户集合起来，去重并排序，然后拼成一个以逗号分隔的字符串。排序并不是业务必须的，但可以让中间结果看起来更规整，第二阶段做两两组合时也更容易复现问题。

import java.io.IOException; import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Set; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; public class FriendIndexReducer extends Reducer<Text, Text, Text, Text> { @Override protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { Set<String> userSet = new HashSet<>(); for (Text val : values) { userSet.add(val.toString()); } List<String> users = new ArrayList<>(userSet); Collections.sort(users); context.write(key, new Text(String.join(",", users))); } }

3.2 第二阶段Mapper与Reducer

第二阶段Mapper输入是第一阶段Reducer的输出，key是某个被关注者，value是关注者列表字符串。Mapper做两两组合时，要先把关注者列表拆成数组，再套两个循环。为了不出现“user001:user002”和“user002:user001”两种键，组合前先按字符串顺序排一下：

import java.io.IOException; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; public class CommonFriendMapper extends Mapper<Text, Text, Text, Text> { @Override protected void map(Text key, Text value, Context context) throws IOException, InterruptedException { String friend = key.toString(); String[] users = value.toString().split(","); for (int i = 0; i < users.length; i++) { for (int j = i + 1; j < users.length; j++) { String a = users[i]; String b = users[j]; String pairKey = a.compareTo(b) < 0 ? a + ":" + b : b + ":" + a; context.write(new Text(pairKey), new Text(friend)); } } } }

用户数量是n时，关注者列表两两组合会产生n乘n减1再除以2个中间键值对。一个拥有十万关注者的用户，单独就能产生接近五十亿条中间记录，这个数量级也直接决定了第二阶段的数据倾斜风险，后面我会专门讲这个问题。

第二阶段Reducer把同一个用户对下所有的共同好友收集起来，同样做去重和排序，再以逗号分隔输出：

import java.io.IOException; import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Set; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; public class CommonFriendReducer extends Reducer<Text, Text, Text, Text> { @Override protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { Set<String> friendSet = new HashSet<>(); for (Text val : values) { friendSet.add(val.toString()); } List<String> commonFriends = new ArrayList<>(friendSet); Collections.sort(commonFriends); context.write(key, new Text(String.join(",", commonFriends))); } }

3.3 Driver配置与作业串联

因为这里有两个MapReduce作业，而且第二个作业依赖第一个作业的输出，所以Driver里需要串行提交两个Job。第一个作业的输出目录不能和输入目录重合，第二个作业的输入目录直接指向第一个作业的输出目录即可。

import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class CommonFriendDriver { public static void main(String[] args) throws Exception { if (args.length < 3) { System.err.println("Usage: CommonFriendDriver <input> <job1Out> <job2Out>"); System.exit(1); } Configuration conf = new Configuration(); Path inputPath = new Path(args[0]); Path job1Out = new Path(args[1]); Path job2Out = new Path(args[2]); Job job1 = Job.getInstance(conf, "BuildFriendIndex"); job1.setJarByClass(CommonFriendDriver.class); job1.setMapperClass(FriendIndexMapper.class); job1.setReducerClass(FriendIndexReducer.class); job1.setOutputKeyClass(Text.class); job1.setOutputValueClass(Text.class); FileInputFormat.setInputPaths(job1, inputPath); FileOutputFormat.setOutputPath(job1, job1Out); if (!job1.waitForCompletion(true)) { System.exit(1); } Job job2 = Job.getInstance(conf, "FindCommonFriends"); job2.setJarByClass(CommonFriendDriver.class); job2.setMapperClass(CommonFriendMapper.class); job2.setReducerClass(CommonFriendReducer.class); job2.setOutputKeyClass(Text.class); job2.setOutputValueClass(Text.class); FileInputFormat.setInputPaths(job2, job1Out); FileOutputFormat.setOutputPath(job2, job2Out); System.exit(job2.waitForCompletion(true) ? 0 : 1); } }

这里有一个配置层面的细节容易被忽略：如果两个作业的Mapper输出key或value类型不一致，必须单独设置setMapOutputKeyClass和setMapOutputValueClass。我这个案例里两个阶段的输出键值都是Text，所以Driver里的配置比较简洁，但如果你在第二阶段改用了自定义对象，就一定要显式区分Map端和Reduce端的输出类型。

3.4 键类型选择：用Text还是自定义Writable

第一版代码里我老老实实用Text做键，简单直接，也足够应对百万级用户的数据量。如果数据量继续往上走，想利用自定义的WritableComparable来减少排序时的字符串比较开销，会在性能上有提升，但编码复杂度也会上去。

现实中我倾向于先跑通Text版本，确认业务结果正确后再考虑优化键类型。因为共同好友这个需求真正的瓶颈往往不在键比较，而在shuffle数据量和热点key的分布。过早优化键类型，可能花了很大力气却只缩短了几分钟运行时间，性价比不高。

4. 跑批实录：数据倾斜、重复值与类型陷阱的排查过程

4.1 第二阶段Reducer长尾：完整排查链路

第一版作业跑全量数据时，我注意到一个非常典型的故障现象——第二个作业里绝大多数Reducer几十秒就跑完了，但有一个Reducer跑了十几分钟还在磨。集群资源没有异常，输入数据量也不大，明显是数据倾斜。

为了定位具体是哪个key倾斜，我先在作业页面上查看每个Reducer的输入记录数，发现某个Reducer的输入记录数比其他Reducer高出几个数量级。顺着这个线索去查第二阶段的中间输出，果然发现了一个拥有百万级关注者的大号用户。这个用户作为倒排key时，值列表特别长，两两组合产生的中间记录数呈平方级增长，直接把一个Reducer打爆了。

定位到热点key之后，我做了两层处理。第一层，观察这个key对应的组合任务是否可以再拆分，比如按第一个用户id的范围做二次分区，把一个大key拆到多个Reducer上；第二层，如果业务上没有强需求要一次性算出全量两两关系，可以先把大key单独摘出来，存成独立文件，用一个单独的MapReduce作业去跑，避免拖累主链路。

4.2 重复好友导致的结果翻倍

调试过程中我还碰见一个比较隐蔽的问题：输出结果里某些用户对的共同好友出现了重复值。排查后发现，源头有两个，一个是输入行本身包含重复好友，另一个是业务上“关注”和“好友”数据出现交叠，同一个关系在两条不同的上游记录里各出现了一次。

解决思路是在每个阶段的Reducer里都做一次Set去重。第一阶段把同key下的用户集合去重，第二阶段把同用户对下的共同好友集合去重。这样虽然增加了一点内存开销，但能保证最终输出每行里不会出现同一个好友被拼两次的问题。后来我还专门写了一个校验程序，对输出结果逐行检查是否有重复项，再抽样和线上数据库比对，确保两边口径一致。

4.3 字符集、分隔符与序列化的雷区

另一个容易翻车的地方是字符集。Hadoop Text的默认字符集是UTF-8，但如果上游是GBK编码的文件，直接toString后再split，中文id会出现乱码，分组键对不上，结果稀碎。处理方式是在Configuration里显式设置mapreduce.map.output.compress.codec等参数时连带检查输入编码，或者要求上游统一输出UTF-8。

分隔符也有讲究。有些上游数据会用分号、制表符甚至多个空格混排。我的习惯是先用indexOf(':')按表现最可靠的一个分隔符切第一刀，右侧好友列表再统一按逗号切。先用split(":")的话，万一行里有其他冒号，数组长度会变，逻辑就容易出错。

序列化方面，Text对象在循环里反复new会造成大量对象创建开销。如果单个Mapper处理的数据量很大，可以重复使用同一个Text实例，调用set方法更新内容。我在第二版优化里就改了这一点，GC时间明显下降。

4.4 小文件导致第二阶段Mapper风暴

第一阶段作业输出结果时，如果Reducer数量设置得太多，会产生大量小文件。这些文件直接作为第二阶段输入，会让第二阶段启动大量根本没多少数据可算的Mapper，白白消耗资源。

我当时把第一阶段的Reducer数量从默认值调到了和输入数据规模匹配的一个值，同时开启输出合并压缩。这里有个经验值可以参考：让每个Reducer输出文件大小大致在64MB到128MB之间，既不至于文件太碎，也不会因为单文件过大导致后续任务等待时间变长。如果你的集群配置了HDFS的归档或合并机制，也可以用工具做一次小文件合并，再交给第二阶段。

5. 从共同好友功能延伸出去：关系推荐与替换方案的取舍

5.1 共同好友结果如何变成推荐候选

共同好友两张表算出来后，最常见的下游用途就是关系推荐。产品里通常会给用户推荐“你可能认识的人”，共同好友数量是最重要的一个信号。两两关系表里每个用户对对应一个共同好友列表，列表长度就是一个天然的特征值。可以按照共同好友数量降序排序，过滤掉已经建立直接关系的用户对，再配合时间衰减、地域等特征，生成一份推荐候选集。

这个Case里还可以衍生出一个更细分的需求：给指定用户推荐好友时，只取其所在行，不需要全量扫描。由于输出格式是user001:user002 -> user003,user012，如果用户id固定，直接按前缀过滤即可。如果希望按某个共同好友倒查所有用户对，那就需要把结果表再做一次转置，变成friend -> userPair的反向索引。

5.2 内存和IO优化思路

大规模跑批时，IO和内存是主要瓶颈。第一阶段的Mapper读取一行就能输出多对键值，shuffle数据量往往比原始数据大很多倍。可以考虑在Mapper端加Combiner，先对每个friend下的user做一次本地去重，减少网络传输。

第二阶段的热点key问题，还可以通过引入自定义Partitioner改善。比如按照用户对中第一个用户id的哈希值做更细粒度的分区，避免所有大组合都落在同一个Reducer上。不过需要说明的是，自定义Partitioner只是在分配层面缓解，如果单条中间结果本身过大，还是需要通过拆分大key或者修改业务口径来根治。

5.3 如果换用Hive或Spark，思维要变在哪里

如果团队里已经有Hive或Spark环境，其实不需要在MapReduce层面硬写两个作业。Hive可以把第一阶段写成一次INSERT OVERWRITE，用collect_list做聚合，第二阶段再用explode或lateral view把所有用户对拆开。Spark更直接，用groupBy得到每个好友的关注者列表，再在内存里用combinations生成用户对，最后groupBy一次对用户对聚合即可。

但换工具不等于换思想。底层的核心仍然是把“用户到好友列表”翻成“好友到用户列表”，再在这个倒排结构上做两两组合。MapReduce版本虽然写起来繁琐，但对理解分布式计算的shuffle、分区和倾斜问题帮助很大。我这一趟跑下来最大的体会是：这类两两关系的分布式计算，真正难的不是API，而是怎么设计键才能让需要汇聚的数据落到同一个Reducer上。

最后再分享一个我自己的小习惯：每次跑完这种多阶段的MapReduce任务，我都会把中间产物留一份在HDFS上，不要急着删除。后续排查数据问题时，这些中间文件是最好的线索来源。等确认所有下游消费方都稳定了，再定期清理也不迟。