最近在做社交关系数据的离线统计分析接到一个特别朴素的需求在千万级用户的好友关系列表里找出任意两个用户之间的共同好友输出成一张两两关系表。产品那边只给了这么一句话但真正动手之后才发现从需求到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 MapperLongWritable, 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(,); SetString 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 ReducerText, Text, Text, Text { Override protected void reduce(Text key, IterableText values, Context context) throws IOException, InterruptedException { SetString userSet new HashSet(); for (Text val : values) { userSet.add(val.toString()); } ListString 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 MapperText, 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 ReducerText, Text, Text, Text { Override protected void reduce(Text key, IterableText values, Context context) throws IOException, InterruptedException { SetString friendSet new HashSet(); for (Text val : values) { friendSet.add(val.toString()); } ListString 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上不要急着删除。后续排查数据问题时这些中间文件是最好的线索来源。等确认所有下游消费方都稳定了再定期清理也不迟。