企业官网建设流程全解析

订单表和用户表做关联，这事儿放在关系型数据库里就是一句left join的事，搬到 Hadoop 上却能让不少人卡上好几天。我第一次动手写 MapReduce 的关联逻辑时，也把 ReduceJoin 想得太简单了，觉得无非是在 Reducer 里套两层 for 循环。结果跑出来的结果集比预期大了一倍多，同时还有一个 Reduce 任务卡在 99% 一动不动。后来才慢慢明白，ReduceJoin 这个名字里的重点根本不在 “Join”，而在 “Reduce”——真正的战场是 Shuffle，是分组之后同一 key 下数据的组织方式，是大小表在内存里的摆放策略。这篇就围绕 MapReduce 里 ReduceJoin 这个经典案例，把我在实际写代码、调倾斜、查日志过程中攒下来的东西完整摊开讲一遍。适合刚学完 WordCount、准备上手真正业务逻辑的 Hadoop 初学者，也适合写了一阵子 MR 但一直被数据倾斜和数据膨胀困扰的同学。看完你至少能搞清楚三件事：标签该怎么打、笛卡尔积怎么写不炸内存、倾斜该怎么缓解。

1. 从一句 SQL 的 left join 说起：ReduceJoin 究竟在哪一步完成连接

1.1 分布式环境下的 Join 和单机数据库的 Join 完全不是一回事

单机数据库做 Join，底层可以是 nested loop，可以是 hash join，也可以是 sort-merge join。不管哪一种，前提都是两张表的数据在同一个进程能访问到的范围内，索引、缓冲区、执行计划优化器这些东西都在帮你。到了 MapReduce 里，这个前提直接没了：两张表的数据被切成若干个 block 散落在不同机器上，任何一个 Mapper 手里都只有一小片数据，它根本不知道自己这片数据对应另一张表的哪些行。这就是为什么 MapReduce 必须把 “连接” 这个动作往后推，推到 Reduce 阶段——因为只有经过 Shuffle，相同 key 的数据才会被聚到同一个 Reducer 里，连接才有可能发生。

理解这一点非常重要，它直接决定了 ReduceJoin 的整体形态：Map 端不做任何连接，只负责给每条记录贴一个来源标签，然后按关联字段作为 key 输出；Shuffle 阶段负责把相同 key 的记录汇聚到一起；Reduce 端拿到同一个 key 下的两组数据（来自表 A 的和来自表 B 的），再做组装。整个过程可以类比成这样一个场景：两个仓库各自把货物贴上 “来自一号仓” 和 “来自二号仓” 的标签，然后按货物编号统一送到分拣中心，分拣员看到同一个编号下同时有一号仓和二号的货，才把这两件捆在一起发出去。分拣中心就是 Reduce，贴标签就是 Map 端的核心动作。

这里面有个关键推论：Map 端输出的 key 必须是关联字段，而不是原始表的主键。这一点新手特别容易搞错。比如订单表用 order_id 做主键，用户表用 user_id 做主键，你要按 user_id 关联，那 Mapper 输出的 key 就应该是 user_id，两张表都得这么输出。如果你图省事直接拿整行数据当 key，那 Shuffle 之后每个 key 下就只有一条记录，Reduce 里压根凑不出对，结果就是一个空结果集，排查起来还特别费劲。

1.2 一次完整的数据流走一遍，心里才有底

拿一个具体的例子来走一遍流程。假设有两份数据：订单表order.txt，字段是order_id, user_id, amount；用户表user.txt，字段是user_id, name, city。目标是输出order_id, user_id, name, city, amount，也就是把用户名和城市补到订单上。

第一步，用FileInputFormat.addInputPaths同时把两个目录加进去（实际项目里通常是两个不同的 HDFS 路径，比如/data/order/20240101和/data/user/dt=20240101）。这时候要特别注意，两个目录下文件的字段结构是不一样的，Mapper 必须能分辨出当前处理的这一行来自哪张表。

第二步，Mapper 在setup()里通过FileSplit拿到当前分片对应的文件路径，据此判断数据来源，然后在map()里给每条记录打上order或user的标签，输出<user_id, "order\t订单字段...">或<user_id, "user\t用户字段...">。

第三步，Shuffle。这一步是黑盒，但你必须知道它干了什么：按 key 的 hash 分区、在内存缓冲区排序、溢写到磁盘、最后归并。相同的 user_id 一定会落到同一个 Reducer，并且同一个 key 下的所有 value 会以迭代器的形式一次性交给reduce()方法。

第四步，Reducer 遍历这个迭代器，把order标签的记录存到一个集合，user标签的记录存到另一个集合，然后对两个集合做笛卡尔积，逐条写出。正常业务里一个用户对应多个订单、一个订单只对应一个用户，所以笛卡尔积的规模其实等于订单数，不会爆炸。但如果关联字段本身有重复（比如用户表里同一个 user_id 出现了两条），那结果就会翻倍，这就是我开头说的 “结果集大了一倍多” 的真实原因。

把这条链路在脑子里过一遍，你就会发现 ReduceJoin 的所有坑几乎都集中在这几个点上：标签打得对不对、key 选得对不对、内存放得放不下、倾斜严不严重。后面几节就沿着这条线逐个拆。

2. Map 端的标签设计：一行代码背后藏着的序列化与分片判断

2.1 为什么标签不能省，以及三种常见的实现方式

有人会想，能不能不打标签，让 Reducer 自己根据字段数量判断这条记录来自哪张表？这在字段数不同的情况下勉强能跑，但只要两张表字段数一样，或者上游数据有空列导致 split 之后长度变化，立刻就会出错，而且这种错误是静默的，不会抛异常，只会让结果悄悄错掉。所以标签必须显式打，而且要打得足够醒目，最好是一个不可能和业务数据冲突的前缀。

实现上一般有三种路子。第一种最土也最直观：直接在 value 字符串前面拼一个标记，用\t或者某个特殊分隔符隔开，Reducer 端split之后取第一个元素判断来源。第二种是定义一个自定义的Writable类，把标签和字段都作为成员变量，实现write()和readFields()，好处是类型安全、可读性好，代价是每次输出都要多做一次序列化对象创建。第三种是把标签编到 key 里，比如输出<user_id + "#" + flag, 数据>，Reduce 阶段在分组比较器里把#后面的部分剥掉——这种写法可以用来控制同一 key 下两类数据的到达顺序，属于进阶玩法，一般场景用不着。

我在教学和实际项目里基本都用第一种，理由很实在：调试方便。你直接在hdfs dfs -cat part-r-00000的时候就能肉眼看出每条数据来自哪张表，出问题的时候省掉大量时间。第二种适合字段特别多、需要强类型约束的场景，但要注意Writable的实现里write()和readFields()的字段顺序必须严格一致，写反了不会报错，只会让你读出乱码，这是新手最容易栽的地方。

2.2 用 FileSplit 判断来源，以及它失效的几种情况

判断当前行来自哪个文件，最常用的方式是重写 Mapper 的setup()：

public class ReduceJoinMapper extends Mapper<LongWritable, Text, Text, Text> { private String sourceTag; private Text outKey = new Text(); private Text outValue = new Text(); @Override protected void setup(Context context) throws IOException, InterruptedException { // 拿到当前分片，强转成 FileSplit 后取文件路径 FileSplit split = (FileSplit) context.getInputSplit(); String fileName = split.getPath().getName(); if (fileName.startsWith("order")) { sourceTag = "order"; } else if (fileName.startsWith("user")) { sourceTag = "user"; } else { throw new IOException("无法识别的输入文件: " + fileName); } } @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); if (line == null || line.trim().isEmpty()) { return; } String[] fields = line.split("\t"); if ("order".equals(sourceTag)) { if (fields.length < 3) { context.getCounter("JOIN", "BAD_ORDER_LINE").increment(1); return; } outKey.set(fields[1]); // user_id 作为关联键 outValue.set("order\t" + fields[0] + "\t" + fields[2]); } else { if (fields.length < 3) { context.getCounter("JOIN", "BAD_USER_LINE").increment(1); return; } outKey.set(fields[0]); outValue.set("user\t" + fields[1] + "\t" + fields[2]); } context.write(outKey, outValue); } }

这段代码有两个细节值得说。第一个是setup()里的异常处理——如果文件名既不以 order 开头也不以 user 开头，我选择直接抛异常让任务失败，而不是默默跳过。原因很现实：一旦上游多塞了一个文件进来，静默跳过会让你以为任务成功了，等你发现数据少了再去回溯，成本高得多。宁可让它快速失败。

第二个是计数器。我在跳过脏数据的时候顺手把计数器加一，这样任务跑完在 Web UI 或者命令行日志里就能看到JOIN.BAD_ORDER_LINE有多少条。这个习惯非常重要，它相当于给数据质量装了一个仪表盘。我见过太多人跳过脏数据什么都不记，结果下游发现数据量对不上，又回头重跑，白烧几个小时集群资源。

至于FileSplit强转的坑，主要出现在这几种情况：用了CombineFileInputFormat或者NLineInputFormat的时候，getInputSplit()返回的就不是FileSplit了，强转会直接抛ClassCastException；用了自定义的InputFormat同理。还有一种更隐蔽的：本地模式跑的时候没问题，上了集群发现某些分片拿到的路径是临时目录，判断逻辑就崩了。所以如果你确实需要兼容这些情况，稳妥的做法是在setup()里先instanceof判断一下，拿不到路径就退化成按行内容特征判断，虽然丑但能兜底。

2.3 一个被反复踩的坑：迭代器对象复用

这个坑严格说不在 Map 端，但它是 ReduceJoin 代码里最容易出错的地方，提前放在这里讲更连贯。Hadoop 为了减少 GC 压力，在reduce()里传给你的Iterable<Text> values是会复用对象的——你如果直接list.add(value)把引用存起来，最后你会发现整个 list 里全是同一条数据，因为所有引用指向的是同一个被反复覆写的对象。

正确的做法是在遍历的时候立刻把内容转成字符串或者克隆一份：

for (Text t : values) { String record = t.toString(); // 立刻转 String，切断对复用对象的引用 if (record.startsWith("order\t")) { orderList.add(record); } else { userList.add(record); } }

这个坑最恶心的地方在于它不是每次都出问题。数据量小的时候、JVM 还没来得及复用对象的时候，跑出来是对的；一上量就错，而且错得毫无规律。我第一次遇到的时候调了整整一个下午，最后是在reduce()里打日志才发现打印出来的十条记录居然一模一样。从那以后我写任何 MR 代码，只要涉及到把 value 存集合，第一反应就是先toString()。

3. Reduce 端的组装逻辑：先分类再笛卡尔积，顺序不能反

3.1 两类数据该放在哪里，内存怎么算

Reducer 拿到一个 key 下的所有 value，标准做法是准备两个集合，一个存主表（一般是数据量大的那张，比如订单），一个存从表（数据量小、需要被广播的那张，比如用户）。然后对两个集合做双重循环输出。但这里有个内存问题必须提前算清楚：一个 key 下的数据量有多大，决定了你能不能全放内存。

以订单关联用户为例，一个用户对应的订单数通常在几十到几百这个量级，极端的大卖家可能有几万单，但那是少数。用户表这边更简单，一个 user_id 正常只有一条记录。所以内存压力很小。但如果你的关联场景是 “商品关联所有购买过它的订单”，一个爆款商品可能有上百万订单，那把所有订单堆进ArrayList必然 OOM。

我一般的判断标准是：预估单 key 最大记录数 × 单条记录平均字节数，如果超过 Reducer 分配的堆内存的 30%，就不要全量加载了，改成 “小表全加载、大表流式遍历” 的模式。Hadoop 的Iterable迭代器有个特性：它背后是磁盘上的归并结果，你可以一边遍历一边处理，不需要把全部数据读进内存。所以更省内存的写法是把小表（用户表）先全部读进内存，然后对大表（订单表）做流式遍历，在遍历过程中直接输出：

public class ReduceJoinReducer extends Reducer<Text, Text, Text, Text> { private List<String[]> userList = new ArrayList<>(); private Text outKey = new Text(); private Text outValue = new Text(); @Override protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { userList.clear(); List<String[]> orderList = new ArrayList<>(); // 第一遍：把数据按来源分开放 for (Text t : values) { String record = t.toString(); String[] arr = record.split("\t"); if ("user".equals(arr[0])) { userList.add(arr); } else if ("order".equals(arr[0])) { orderList.add(arr); } else { context.getCounter("JOIN", "UNKNOWN_TAG").increment(1); } } // 从表为空直接跳过，对应 inner join 语义 if (userList.isEmpty()) { context.getCounter("JOIN", "ORDER_WITHOUT_USER").increment(orderList.size()); return; } // 笛卡尔积输出 for (String[] order : orderList) { for (String[] user : userList) { outKey.set(order[1]); outValue.set(order[2] + "\t" + user[1] + "\t" + user[2]); context.write(outKey, outValue); } } } }

注意这个写法还是把 orderList 存下来了，因为Iterable只能遍历一次。如果你要做真正的流式处理，就得改成两阶段：第一遍遍历只收集小表，第二遍遍历输出大表——但Iterable不支持重置，所以这条路走不通，除非你在 Mapper 端用二次排序把从表数据排到前面，Reduce 端读到标签变化就知道小表读完了。

这就是二次排序（Secondary Sort）在 ReduceJoin 里的用武之地，属于进阶优化，后面第 6 节会展开说。日常业务里如果单 key 数据量可控，上面这种 “两个 List 全装” 的写法完全够用，代码可读性也最好。

3.2 left join 还是 inner join，语义得自己实现

Hadoop 不会帮你区分 join 类型，你写什么就是什么。上面代码里如果userList.isEmpty()就直接return，那语义就是 inner join——用户表里没有的订单会被丢掉。但业务上经常需要 left join：用户信息缺失的订单也要保留，只是相关字段置空。

实现方式很简单，加一个分支：

if (userList.isEmpty()) { // left join：保留订单，用户字段补空 for (String[] order : orderList) { outKey.set(order[1]); outValue.set(order[2] + "\t" + "NULL" + "\t" + "NULL"); context.write(outKey, outValue); } context.getCounter("JOIN", "ORDER_WITHOUT_USER").increment(orderList.size()); return; }

但这里有个业务判断要注意：NULL到底该写成字符串"NULL"、空字符串，还是干脆不写这个字段？取决于下游怎么消费。如果下游是 Hive 表，建议用空字符串或者\N（Hive 默认的 NULL 表示），别写"NULL"字符串，否则下游统计count(name)的时候会把"NULL"当成一个真实的名字算进去，数据就错了。这种细节看着小，但一旦埋进去，排查起来要翻好几层下游任务。

还有反连接的场景：只想找出用户表里没有的订单（用于数据质量核查）。这时候逻辑反过来，userList非空的反而不输出。这类需求在数据治理里非常常见，比如核查订单表里有多少 user_id 在用户维度表里查无此人。用同样的框架改几行就能实现，这也是 ReduceJoin 比写 SQL 灵活的地方——你可以任意组合输出条件。

3.3 输出格式设计与分区器的小调整

ReduceJoin 输出的 key 用哪个字段，很多人不假思索就用原表主键。但如果后续还要按 user_id 再做一次聚合，用 order_id 当输出 key 就意味着下一次作业还得重新 Shuffle 一遍。所以我的习惯是看下游：如果这个结果马上要被另一个按 user_id 分组的作业消费，那输出 key 就直接用 user_id，让两次作业共用一次分区结果——当然这需要两个作业的 Reducer 数量一致，而且严格来说中间还隔着一次落盘，收益有限，别为了这点优化把代码搞复杂。

真正值得改的是分区器。默认的HashPartitioner按 key 的 hashCode 取模，如果关联字段分布不均（比如某个 user_id 特别高频），就会导致某个 Reducer 分到的数据远超其他。这种情况要么换自定义分区器把热点 key 打散，要么提前做预处理。自定义分区器的写法不复杂，继承Partitioner重写getPartition()就行，但要注意打散之后同一个 key 的数据会分到不同 Reducer，那就没法在这个 Reduce 里做完整 join 了，所以分区器只能解决 “负载均衡” 问题，解决不了 “join 正确性” 问题，这两者得分开看。

4. 数据倾斜：ReduceJoin 绕不开的那道坎

4.1 倾斜是怎么产生的，为什么 ReduceJoin 特别敏感

数据倾斜的本质是 key 分布不均导致某些 Reducer 分到的数据量远大于其他。ReduceJoin 对这个问题格外敏感，因为它通常关联的是业务主键，而业务主键的分布天然就不均匀。用户表里几十个真实用户可能对应了平台上一大半的订单，这些用户的 user_id 就是超级热点，落到哪个 Reducer，哪个 Reducer 就得独自扛完全部数据。

具体表现非常有辨识度：打开 Job 的 Web UI，你会看到 99% 的 Reduce 任务在几分钟内就完成了，剩下那么一两个卡在 66% 或者 99% 一动不动，进度条长时间不变化，Map 阶段早就 100% 了。这时候去看 Counter 里的Reduce input records，会发现各个 Reduce 的输入记录数差了两三个数量级。我第一次见到这个现象时还以为是集群卡了，重启了一遍任务，结果还是一样，纯属浪费资源。

还有一个更隐蔽的表现：任务不是卡住，而是直接失败，报java.lang.OutOfMemoryError: Java heap space或者GC overhead limit exceeded。这种情况是热点 key 下的数据量太大，Reduce 端的两个 List 直接把堆撑爆了。注意这个报错和 “容器被 NodeManager 杀掉” 是两回事，前者是 JVM 内部 OOM，后者是超出了mapreduce.reduce.memory.mb的限制被外部 kill，日志上的关键字不一样，排查方向也不同。

4.2 缓解手段一：调参兜底，先让它能跑完

最直接的手段是给 Reducer 加内存。通过-D mapreduce.reduce.memory.mb=4096这类参数把容器内存提上去，同时把mapreduce.reduce.java.opts里的-Xmx也相应调整，一般是容器内存的 75% 到 80%。这个手段简单粗暴，但有效，适合 “就差一点点就能跑完” 的场景。

第二个参数是mapreduce.reduce.shuffle.input.buffer.percent，控制 Shuffle 阶段用于缓存的内存比例。倾斜严重的时候，热点 Reducer 要拉取的数据量极大，适当调大这个比例能减少溢写次数，加快 Shuffle。默认 0.7，一般不用动，但在确认是 Shuffle 阶段卡住时可以试试。

第三个是增加 Reduce 数量。默认一个 Reducer 处理所有数据，改成-D mapreduce.job.reduces=10，虽然热点 key 还是只会落到其中一个 Reducer，但至少小 key 的数据被分摊了，整体完成时间会明显缩短。这个手段治标不治本，但如果你的目标是 “今晚先把报表跑出来”，它是最省事的选择。

要提醒一句：调参能救急，但别把它当成长期方案。我见过有团队把所有作业的 Reducer 内存统一拉到 8G，理由是 “反正集群资源够”，结果集群整体利用率一塌糊涂，小作业也占着大容器不放，排队时间反而变长。参数该按作业特点分别设，别搞一刀切。

4.3 缓解手段二：给热点 key 加盐，两阶段聚合

真正解决问题的思路是把热点 key 拆开。具体做法分两步走：第一个作业在 Mapper 端给 key 加一个随机前缀（比如0_到9_），这样原本一个热点 key 就被拆成了十个，分散到不同 Reducer 上做局部 join；第二个作业把前缀去掉，再做一次汇总。这就是所谓的 “加盐”。

加盐的代码改动不大，Mapper 里输出 key 的时候加个随机数就行：

// 只对从表（小表）加盐，主表不加，保证每条主表记录能和所有盐值的从表记录匹配上 int salt = new Random().nextInt(10); if ("user".equals(sourceTag)) { outKey.set(salt + "_" + fields[0]); } else { // 主表需要发 10 份，每份对应一个盐值 for (int i = 0; i < 10; i++) { outKey.set(i + "_" + fields[1]); outValue.set("order\t" + fields[0] + "\t" + fields[2]); context.write(outKey, outValue); } }

看到这里你可能会皱眉：主表数据被复制了 10 份，数据量直接膨胀十倍，这划算吗？答案是看情况。如果主表本身不大，而从表有超级热点，那这个交换是值得的；如果主表本身就是海量数据，那加盐反而会雪上加霜。所以加盐之前一定要先搞清楚 “谁是热点”——是主表的某个 key 对应了海量从表记录，还是反过来。搞反了方向，优化会变成劣化。

实际生产中我更常用的是一种折中方案：对超过阈值的 key 单独处理。先用一个抽样作业统计出 key 的频次分布，找出 top N 的热点 key，然后在主作业里把这些 key 的数据单独走一条处理链路（比如写到另一个输出目录，用小规模作业单独跑），剩下的正常 key 走常规 ReduceJoin。这个方案实现复杂度高一些，但对数据量的影响最小，适合对时效和资源都敏感的场景。

5. 本地跑通和集群跑通，中间隔着多少个坑

5.1 本地模式的路径判断为什么会失效

在 IDE 里用LocalJobRunner跑的时候，FileSplit.getPath()返回的路径和你想象的可能不一样。本地模式下输入路径是file:///...，而集群上是hdfs://ns/...。如果你的判断逻辑里写了绝对路径前缀匹配，本地能跑集群就崩。我一般只取getName()拿文件名，不碰完整路径，这样就绕开了协议前缀的差异。

还有一个更隐蔽的差异：本地模式下mapreduce.job.reduces默认是 1，所有 key 都进同一个 Reducer，你根本看不出倾斜问题；一上集群改成多个 Reducer，各种问题才冒出来。所以本地测试通过绝不代表集群能跑通，这一点必须有心理准备。

另一个常见问题是setup()在本地可能被调用多次（每个分片一次，而本地分片划分和集群不同），如果setup()里有副作用（比如写文件、初始化连接池），本地跑着没问题，集群上可能会出现资源泄漏。稳妥的做法是把setup()里的逻辑做成幂等的，或者把资源初始化放在run()里统一管理。

5.2 用 Counter 和日志把问题定位到具体环节

MR 作业出问题的时候，最有效的定位手段其实不是打断点，而是看 Counter。Hadoop 内置了一堆计数器，我最常看这几个：MAP_INPUT_RECORDS和MAP_OUTPUT_RECORDS对比，能看出 Mapper 有没有吞数据；REDUCE_INPUT_GROUPS和REDUCE_INPUT_RECORDS对比，能算出平均每个 key 下的记录数，这个数字直接反映倾斜程度；COMBINE_INPUT_RECORDS和COMBINE_OUTPUT_RECORDS对比，能判断 Combiner 有没有起作用。

自己加的计数器也很有价值。前面代码里那几个JOIN.BAD_ORDER_LINE、JOIN.ORDER_WITHOUT_USER，跑完之后直接看数值就知道数据质量如何。我习惯在关键分支上都埋一个计数器，代价几乎为零，收益是排查问题时不用再猜。

日志方面，context.getLogger()打出来的日志会进到 Job 的日志文件里，不在标准输出。注意别在reduce()里对每条记录都打日志，几百万条记录打下来日志文件能撑爆磁盘，而且会严重拖慢任务。我的做法是只在第一条和最后一条打，或者用if (count % 100000 == 0)这种采样方式。

5.3 常见异常对照表

这张表里每一条都是我或者身边同事真实踩过的，尤其是最后两条，属于 “不报错但结果错” 的类型，危害远大于直接失败。任务失败你至少知道要去查，结果错了可能过了半个月才被业务方发现。

6. ReduceJoin 的适用边界与几种替代思路

6.1 什么时候该果断放弃 ReduceJoin

ReduceJoin 最大的优点是通用：不管两张表多大、怎么分布，只要磁盘够、时间够，它总能算出来。但它的代价也很明确——全量数据都要过一遍 Shuffle，网络传输和磁盘 IO 的成本摆在那。所以判断标准其实很简单：如果有一张表小到能整个塞进每个 Mapper 的内存，那就完全没必要走 Reduce。

这就是 MapJoin（也叫广播 Join）的用武之地。通过job.addCacheFile()把小表分发到每个节点，Mapper 在setup()里一次性读进内存，构建成一个HashMap，然后处理大表的时候直接查表，连 Shuffle 都省了。我做过一个粗略对比，同样一份数据，MapJoin 的耗时大概是 ReduceJoin 的三分之一到五分之一，差距主要来自省掉的 Shuffle 阶段。

小表的判断标准，我的经验是 100MB 以内比较稳妥，超过这个量级就要考虑每个 Mapper 的内存够不够了。因为小表是每个 Mapper 都存一份，如果有 200 个 Map 任务，100MB 的表就意味着 20GB 的额外内存占用，这个账必须提前算。另外要注意，MapJoin 只适合 “大表 join 小表”，两张表都大的时候还是得回到 ReduceJoin。

6.2 半连接与 BloomFilter 剪枝

还有一种中间方案叫半连接（Semi Join）：先用一个作业从大表里提取出所有出现过的关联 key，去重后作为 “小表” 广播到下一个作业，第二个作业的 Mapper 在处理从表时，先判断这个 key 在集合里存不存在，不存在直接过滤掉。这样能大幅减少参与 Shuffle 的数据量，尤其适合从表里大部分数据都关联不上的场景（比如用户表有 500 万，但只有 80 万用户下过单）。

更省内存的做法是用 BloomFilter 替代 HashSet 做这个判断。BloomFilter 的优点是空间效率极高，几百万个 key 可能只占几 MB，代价是存在假阳性——它可能把不存在的 key 误判为存在。但对 join 来说，假阳性只会导致多保留一些最终会被过滤掉的数据，不会漏数据，所以是安全的。这个技巧在处理海量维表的时候特别有用，我一般会在文档里写清楚 “BloomFilter 只用于剪枝，最终 join 仍需完整逻辑兜底”。

6.3 几种方案的横向对比

这张表不是让你背下来，而是提供一种思考顺序：先看表的大小，再看匹配率，最后看实现成本。大部分人的第一反应都是写 ReduceJoin，这没错，但写完之后花十分钟评估一下有没有更优解，往往能省下后面几小时的运行时间。

6.4 二次排序在 ReduceJoin 里的实际价值

最后说说二次排序。它的核心是在 key 相同的情况下，再按一个辅助字段排序，这样 Reducer 拿到的迭代器里数据就是有序的。用在 ReduceJoin 上，可以约定从表（小表）排在前面、主表排在后面，Reducer 遍历的时候读到第一个主表记录就知道小表已经全部读完了，此时可以停止收集、直接开始输出。这样就实现了真正的流式处理，大表侧不需要全量加载进内存。

实现上要设置三个东西：自定义的WritableComparable组合 key、Partitioner按组合 key 的第一部分分区、SortComparator按完整组合 key 排序、GroupingComparator只按第一部分分组。四个东西各管一件事，任何一个写错结果都不对，而且错了通常不报异常，只是分组分乱了。这也是为什么我把它放在最后讲——它确实能解决大 key 问题，但调试成本很高，属于 “不到万不得已不用” 的手段。我个人的经验是，能用加盐或者拆分热点 key 解决的，就别上二次排序，后者的心智负担太重，接手的人容易看不懂。

写 ReduceJoin 这几年下来，最大的感受是这个案例表面上是教你怎么写 MapReduce，实际上教的是怎么思考数据在分布式环境下的流动方式。标签、key、分组、内存、倾斜，这五个词基本覆盖了所有离线关联作业的核心矛盾。代码本身不长，真正花时间的是跑起来之后盯着 Web UI 看哪个 Reduce 慢、翻 Counter 算平均 key 数据量、然后决定是加内存还是拆 key。我现在写这类作业，习惯先在纸上把两张表的 key 分布大概画一画，估算一下最热那个 key 能有多少条记录，如果超过内存的 30%，就直接不写全量加载的版本了，省得后面返工。另外提醒一句，Iterable对象复用这个坑，只要涉及把 value 存集合就先toString()，养成肌肉记忆，基本不会再翻车。