企业官网建设流程全解析

1. ReduceJoin到底解决了什么问题：从两张表说起

刚接触 MapReduce 那会儿，我一度以为这东西就是写个 WordCount，把单词数一数就完事了。真正让我开始理解分布式计算思想的，是第一个 ReduceJoin 案例——把订单表和产品表按产品 ID 拼在一起。这个需求放在单机 SQL 里就是一句left join的事，但要在 MapReduce 里实现，你会发现它逼着你去想清楚三件很底层的事情：数据怎么分片、键怎么设计、同一批数据怎么保证最终能凑到一起。把这三点想通了，ReduceJoin 也就通了，后面学 MapJoin、学 Hive 的 join 优化、学 Spark 的 shuffle，都是在这套逻辑上做加减法。

ReduceJoin，说白了就是把关联这个动作放到 Reduce 阶段完成。Map 阶段不做任何真实关联，只负责给每条记录打一个来源标签，然后统一把关联键（比如产品 ID）作为输出的 key 吐出去。框架保证同一个 key 的所有记录一定进入同一个 Reduce 任务，Reduce 拿到这个 key 下面的完整记录集合之后，自己按标签把它们分成两堆，再用嵌套循环拼成结果。听起来朴素，甚至有点笨，但它解决的是一个分布式环境下的根本矛盾：两份数据分散在不同的文件、不同的分片、不同的机器上，我没办法在任意时刻随手去"查"另一张表。MapReduce 给出的答案是——先把相同的键聚到一起，再谈关联。

适合谁来读这篇内容？如果你是刚学完 WordCount、正要往"多输入源"方向走的学习者，这篇基本可以把 ReduceJoin 的每一步掰开揉碎给你看；如果你已经在写业务代码，只是对中间那层 shuffle 到底发生了什么一直犯迷糊，那第 2 节和第 4 节应该能对上你的疑问；如果你已经在用 Hive 但没深究过底层的 join 实现，看完第 6 节的对比，你会更清楚为什么大表 join 小表会被优化器自动改写成另一种形态。下面这套代码是我在本地伪分布式和一个小规模集群上都跑过的完整版本，参数、坑点、以及我在调试过程中真实碰到的问题，都会一并写进去。

2. 原理拆解：一条订单记录从Map到Reduce经历了什么

2.1 打标签：Mapper端最重要的一件事

ReduceJoin 的 Mapper 干的事情极少，但不能少。它读一行数据，判断这行数据来自哪张表，然后给这条记录贴上一个"我是谁"的标记——通常就是一个字符串字段，比如order和product。同时，它要把两表中语义相同的那个字段抽出来作为 key：订单表里的pid，产品表里的商品编号，虽然字段名不同，但含义一样，必须统一。

为什么一定要打标签？因为到了 Reduce 端，两条记录都会被序列化成同一种 value 类型（比如自定义的 Bean），字段结构完全一致，你光看内容已经分不清谁是订单、谁是产品了。有人会想偷懒，用"订单 ID 是否为空"来区分，这在样例数据里能跑通，但只要业务上出现脏数据，或者订单 ID 恰好可以为空，判断立刻失效。我踩过这个坑：早期版本我用的是"产品名为空就是订单"的判定，结果产品表里有一行商品名缺失，整条数据被误判成订单，最后关联结果里凭空多出来一堆空订单。加一个显式 flag 字段只多占几个字节，换来的却是逻辑上的确定性，这笔账很划算。

这里还有一个容易被忽略的点：Mapper 怎么知道当前处理的这行数据来自哪个文件？标准做法是在setup()方法里通过context.getInputSplit()强转成FileSplit，再拿getPath().getName()取文件名，用一个成员变量存起来。之所以放在 setup 而不是 map 里，是因为一个 Mapper 实例通常只处理一个分片，文件名在整个 map 生命周期内不变，放 setup 里只取一次，省掉每行都做一次类型转换的开销。这种"能提到 setup 就别放 map"的习惯，在小数据量上看不出差别，上了几千万行就是实打实的收益。

2.2 分区与排序：为什么同一个pid必须落到同一个Reduce

Map 输出之后要经过 shuffle，这段过程是 ReduceJoin 正确性的根基。默认的HashPartitioner会对 key 取哈希再对 Reduce 个数取模，决定这条记录去哪个 Reduce。这一步天然满足了我们的要求：只要 key 相同，哈希值就相同，模出来的分区号也相同，无论这个 pid 来自订单表还是产品表，最终一定进同一个 Reduce 任务。

排序则是第二层保障。Map 端在溢写之前会对 key 做一次排序，Reduce 端在真正调用 reduce 方法之前还会做一次归并排序，效果是相同的 key 挨在一起。于是 Reduce 的 reduce 方法每次被调用时，传进来的Iterable<TableBean>里装的就是"当前这个 pid 对应的全部记录"，一条不多一条不少。这两点合起来，才让"在 Reduce 里做关联"这件事成为可能。

理解了这一层，很多问题就有了答案。比如有人问：Reduce 个数设成 3，会不会出现订单在 1 号 Reduce、产品在 2 号 Reduce，导致关联不上？答案是不会，前提是两次作业用的分区逻辑一致、key 类型一致。反过来，如果你在 Mapper 里输出 key 的类型一会儿是 Text 一会儿是 String，或者两边的pid一个是"01"另一个是"1"，那即使哈希算法一样，取模结果也可能天差地别，关联必然失败。我在排查一个"结果只有订单没有产品名"的问题时，最后定位就是产品表里的编号被 Excel 自动去掉了前导零，01变成了1，两个 key 在分区器眼里就是两个完全不同的东西。这类问题用肉眼查数据很难发现，得靠打印 key 的十六进制或者长度去比对。

2.3 迭代与拼接：Reduce端的笛卡尔积是怎么组装出来的

Reduce 拿到一堆记录了，接下来的动作非常直白：遍历一遍，按 flag 分成两个集合，订单放一个 List，产品名单独存一个变量（因为一个 pid 在产品表里理论上只有一条）。然后再遍历一遍订单集合，把产品名塞进每条订单，输出。

产品名为什么不用 List 而用变量？因为关联关系是一对一的。如果产品表里有重复的 pid（比如商品改名保留了历史记录），那按变量存就只会保留最后一条，这是一种取舍。想保留全部的话把产品名也放进 List，最后做双层循环输出笛卡尔积——但这会让结果行数膨胀，多数业务场景不需要。我在实际项目里遇到过维表有重复的情况，处理办法不是改 Reduce 逻辑，而是在上游先用 MapReduce 做一次去重，把维表规整成唯一键，这样 join 端就永远是干净的。把数据质量问题挡在 join 之前，比在 join 里面做兼容要省事得多。

另一件必须说清楚的事：Reduce 方法接收到的Iterable不是普通的 ArrayList，Hadoop 为了省内存，在迭代过程中反复复用同一个对象实例，只是每次把新记录的字段填充进去。这就意味着，如果你把遍历得到的对象直接add进 List，最后 List 里所有元素都会指向同一个对象，内容全都变成最后一条记录的值。我第一次写这个案例时就是犯了这毛病，跑出来的结果是"每条订单的产品名都一样"，排查了半天才反应过来。正确做法是每次取出来之后new一个副本，把字段逐个复制过去，或者用工具类做属性拷贝。这个坑几乎每个学 ReduceJoin 的人都会踩一次，后面第 5 节我会再详细说。

2.4 为什么ReduceJoin里绝对不能开Combiner

Combiner 的本质是"在 Map 端提前做一次小规模的 Reduce"。对于求和类操作（比如 WordCount 的计数），这个优化完全等价，跑一百遍结果都一样。但 join 不是可结合的运算。

设想一下：某个 pid 在某个 Map 分片里只出现了订单记录，没出现产品记录。如果开了 Combiner，它在本地就把这批订单"关联"了一遍，发现没有产品名，要么输出空名字，要么直接丢弃，然后才把结果发给 Reduce。等到真正 Reduce 的时候，产品记录可能从另一个分片过来，但订单已经被提前处理坏了，拼不回去了。结果就是部分数据丢失或者产品名全空，而且这种错误不会报异常，只会安静地输出错误结果，最难排查。

所以我的做法很干脆：ReduceJoin 场景下不调用job.setCombinerClass()，一行都别写。有人会说那性能不就浪费了？确实浪费了一部分网络传输，但换来的结果是正确的。真要优化，方向是减少 Map 输出的体积（比如 Bean 里只留必要字段，不塞整个原始行），或者干脆换 MapJoin，而不是在 Combiner 上动歪脑筋。这个取舍很明确：正确性永远优先于那点带宽。

3. 完整实现：订单-产品关联的代码与逐行注释

3.1 数据准备与分隔符约定

我用的样例数据是两份纯文本，用制表符\t分隔。订单表每行三列：订单 ID、产品 ID、数量；产品表每行两列：产品 ID、产品名称。

order.txt 1001 01 1 1002 02 2 1003 03 3 1004 01 4 product.txt 01 显示器A 02 键盘B 03 鼠标C

分隔符这件事值得单独说一句。很多教程用逗号，看着舒服，但真实业务数据里产品名经常带逗号（"某某牌, 大容量"），一split就把列数打乱了，紧接着fields[2]直接数组越界。制表符的冲突概率低得多，而且 Hive 的默认分隔符也是它。如果数据源已经固定是逗号，那就老老实实用带引号处理的 CSV 解析库，别自己手写split(",")。还有个小细节：split的第二个参数建议传-1，这样行尾的空列会被保留，否则"1001\t01\t"切出来只有两个元素，取第三个又炸了。至于表头，样例数据我没有加，如果需要跳过，最简单的做法是在 map 里判断line.startsWith("订单ID")就 return，别去折腾什么偏移量，跨分片的时候偏移量方案很容易漏掉。

3.2 自定义Bean：序列化字段顺序必须两端一致

Hadoop 的 value 需要能在网络上传输、能落盘，所以必须实现Writable接口，自己写write和readFields。这里最容易出问题的地方是两个方法的字段顺序必须严格对应，顺序错一个，读出来的数据就会错位——数字被读成字符串、名字变成订单号，而且不报错。我见过有人在 Bean 里加了个字段，改了write忘了改readFields，跑出来所有金额都是 0，查了一下午。

package com.demo.join; import org.apache.hadoop.io.Writable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; public class TableBean implements Writable { private String orderId; private String pid; private int amount; private String pname; private String flag; public TableBean() { } @Override public void write(DataOutput out) throws IOException { out.writeUTF(orderId == null ? "" : orderId); out.writeUTF(pid == null ? "" : pid); out.writeInt(amount); out.writeUTF(pname == null ? "" : pname); out.writeUTF(flag == null ? "" : flag); } @Override public void readFields(DataInput in) throws IOException { orderId = in.readUTF(); pid = in.readUTF(); amount = in.readInt(); pname = in.readUTF(); flag = in.readUTF(); } @Override public String toString() { return orderId + "\t" + pname + "\t" + amount; } // getter / setter 省略 }

注意writeUTF前面那一堆三目判断。writeUTF(null)会直接抛空指针，而 ReduceJoin 里注定有一半字段是空的（订单没有产品名，产品没有订单号），如果不做兜底，第一次跑就会崩。另外要提供一个无参构造器，框架反序列化的时候靠反射创建实例，没有无参构造会报InstantiationException。writeUTF底层用的是 modified UTF-8，写中文完全没问题，不要被某些老文章误导去用什么writeBytes，那个截断字符会让你怀疑人生。

3.3 Mapper：靠文件名区分数据来源

public class JoinMapper extends Mapper<LongWritable, Text, Text, TableBean> { private final Text outKey = new Text(); private final TableBean bean = new TableBean(); private String fileName; @Override protected void setup(Context context) throws IOException, InterruptedException { FileSplit split = (FileSplit) context.getInputSplit(); fileName = split.getPath().getName(); } @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString().trim(); if (line.isEmpty()) { return; } String[] fields = line.split("\t", -1); if (fileName.startsWith("order")) { if (fields.length < 3) { return; } bean.setOrderId(fields[0]); bean.setPid(fields[1]); bean.setAmount(Integer.parseInt(fields[2])); bean.setPname(""); bean.setFlag("order"); outKey.set(fields[1]); } else { if (fields.length < 2) { return; } bean.setOrderId(""); bean.setPid(fields[0]); bean.setAmount(0); bean.setPname(fields[1]); bean.setFlag("product"); outKey.set(fields[0]); } context.write(outKey, bean); } }

Mapper 的泛型是<LongWritable, Text, Text, TableBean>，输入 key 是行偏移量（我们不用，但必须声明），输出 key 是 Text 类型的产品 ID。这里有个小优化点：outKey和bean都声明成成员变量而不是在 map 里 new，因为一个 Mapper 会处理成千上万行，每行 new 两个对象会给 GC 造成不小的压力。Hadoop 的context.write会立刻把对象序列化写出去，所以复用是安全的。

至于用文件名判断来源，input目录下两个文件叫什么名，Mapper 就认什么名。这就要求上传到 HDFS 之前别手贱改成data1.txt、data2.txt。如果实在要改，那就把判断条件写成fileName.contains("order")，或者用MultipleInputs给不同路径指定不同的 Mapper，后者更规范，但代码要多写一个类，学习阶段没必要。

3.4 Reducer：对象复用的坑与拷贝

public class JoinReducer extends Reducer<Text, TableBean, Text, TableBean> { private final Text outKey = new Text(); private final TableBean outBean = new TableBean(); @Override protected void reduce(Text key, Iterable<TableBean> values, Context context) throws IOException, InterruptedException { List<TableBean> orderList = new ArrayList<>(); String pname = ""; for (TableBean bean : values) { if ("order".equals(bean.getFlag())) { TableBean copy = new TableBean(); copy.setOrderId(bean.getOrderId()); copy.setPid(bean.getPid()); copy.setAmount(bean.getAmount()); copy.setPname(""); copy.setFlag("order"); orderList.add(copy); } else { pname = bean.getPname(); } } for (TableBean order : orderList) { outBean.setOrderId(order.getOrderId()); outBean.setPid(key.toString()); outBean.setAmount(order.getAmount()); outBean.setPname(pname); outBean.setFlag("order"); outKey.set(order.getOrderId()); context.write(outKey, outBean); } } }

那段看着有点笨的手工字段复制，就是防第 2.3 节说的对象复用问题。有人用BeanUtils.copyProperties或者反射工具类，确实简洁，但在 Reduce 里每条记录都反射一次，性能会打折，数据量大的时候这部分开销不可忽略。手写四个 setter，看着啰嗦，跑起来最省。如果你嫌麻烦，可以在 Bean 里加一个copyFrom(TableBean other)方法，把赋值收进去，Reduce 里调一行，兼顾整洁和效率。

另外，最终输出的 key 我用了订单 ID，value 是整个 Bean，toString只输出订单号、产品名、数量三列，其他字段不打印。这样输出文件干净，用cat一看就懂。如果你的下游还要继续加工，那就把 pid 也放进toString，保持信息完整。

3.5 Driver：Reduce个数到底设几个

public class JoinDriver { public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "reduce-join-demo"); job.setJarByClass(JoinDriver.class); job.setMapperClass(JoinMapper.class); job.setReducerClass(JoinReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(TableBean.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(TableBean.class); // 不设置 Combiner，是故意为之 // job.setCombinerClass(...) 千万别加 FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); job.setNumReduceTasks(Integer.parseInt(args[2])); boolean ok = job.waitForCompletion(true); System.exit(ok ? 0 : 1); } }

Reduce 个数通过第三个参数传进去，方便我对比不同设置下的耗时。这里有个新手常犯的错误：Map 输出和最终输出的类型必须分别声明。setMapOutputKeyClass/setMapOutputValueClass管的是 Mapper 输出，setOutputKeyClass/setOutputValueClass管的是最终结果。类型不一致的时候（比如 Mapper 输出 Text，Reducer 输出自定义 Bean），少声明一组，框架会在运行时报类型转换异常，而且报错信息往往指不到真正的位置。我习惯是两组都显式写上，哪怕它们一样，也让代码自己说清楚意图。

还有个隐式要求：输出目录不能已存在。反复调试的时候，每次跑之前要么换个目录名，要么先手动删掉。小技巧是在本地调试脚本里加一句判断，存在就先删，省得每次跑都报FileAlreadyExistsException。

4. 跑得快才叫跑通：ReduceJoin的调优实操

4.1 Reduce个数与输出小文件

Reduce 个数直接决定输出文件数和单任务压力。设成 1，所有 key 都进一个 Reduce，结果是一个文件，看着整齐，但并发度为零；设成 3，输出三个part-r-00000到part-r-00002，每个文件里只包含落到对应分区的数据。注意这个划分是按 key 哈希来的，不是按数据量均分，所以三个文件大小可能差好几倍。

怎么定？我的经验是：数据量在百万行级别，Reduce 给 2 到 4 个就够，再给多也没用，因为单个 Reduce 处理的数据本来就不大，任务启动和调度的开销反而占比更高。真正的生产环境，Reduce 个数一般按"总数据量 / 单个 Reduce 期望处理量"估算，再留一点余量。但 ReduceJoin 有个特殊性——维表（产品表）会被完整地分发到每个 Reduce 上，Reduce 越多，维表被复制的份数越多。如果维表很小无所谓，如果维表不小，Reduce 个数反而要克制。

小文件问题也顺手说一下。如果作业只是中间环节，输出马上要被下一个作业读，那多个小文件会让下一个作业的 Map 数暴涨（一个小文件一个分片）。常见处理是在后面接一个合并任务，或者干脆把 Reduce 设为 1 用于小数据量的验证。我在调试阶段基本都用 1 个 Reduce，跑通了、逻辑对了，再调上去看性能变化。

4.2 数据倾斜：识别方法与三种缓解手段

ReduceJoin 最容易翻车的地方不是代码写错，而是数据分布不均。想象一下产品表里有个"通用配件"的 ID，全站 80% 的订单都挂在这个 ID 上，那么这个 key 对应的 Reduce 要处理几千万条订单记录，别的 Reduce 几十秒就完事了，它还在那儿转，整个作业的耗时被这一个人拖死。

怎么识别？看任务的进度页面，如果某个 Reduce 卡在 66% 或者 99% 很久不动，而其他 Reduce 早就 100%，基本就是它。另一个信号是 reduce 方法内部打印的计数器——给每个 key 记一下 values 的条数，跑完看最大值和平均值的差距，差十倍以上就该警惕了。

缓解手段我实际用过三种。第一种是给热点 key 加随机前缀，在 Mapper 里判断如果是那个热点 ID，就在后面拼_0到_9的随机数，Reduce 端把前缀去掉再聚合。这招对求和类有效，但 join 场景下会让同一个 pid 拆到多个 Reduce，关联就散了，所以得配上"维表也复制十份"的操作，复杂度上来了，谨慎用。第二种是两次作业法：先跑一个作业把热点 key 的数据单独抽出来，用 MapJoin 处理，剩下的走正常 ReduceJoin，最后合并结果。第三种最省事——把大表加个过滤，把热点维度的数据剔除，单独处理，很多时候业务上就能接受。

4.3 自定义分区器解决热点key

如果热点是有限的几个 key，最简单的做法是自己写个Partitioner，把这些 key 直接指定到某个空闲的 Reduce 上，其他 key 走默认哈希。

public class HotKeyPartitioner extends Partitioner<Text, TableBean> { @Override public int getPartition(Text key, TableBean value, int numPartitions) { String pid = key.toString(); if ("01".equals(pid)) { return 0; } return (pid.hashCode() & Integer.MAX_VALUE) % numPartitions; } }

这段代码的关键是& Integer.MAX_VALUE，把负数哈希值转成正数。不加这一句，hashCode()为负的 key 会算出负数分区号，直接抛IllegalArgumentException: Illegal partition，报错信息还不告诉你哪个 key 出的问题，第一次遇到会非常懵。分区器写完别忘了在 Driver 里job.setPartitionerClass(HotKeyPartitioner.class)，并且 Reduce 个数至少为 2，否则所有 key 都只会去 0 号分区。

4.4 溢写与内存参数

Reduce 端把订单记录攒进 ArrayList，这个 List 是堆内存里的对象，条数一多就可能 OOM。默认的 Reduce 堆内存往往不够看。调的方式是在mapred-site.xml里改mapreduce.reduce.java.opts，把-Xmx提上去，比如-Xmx2048m。但更根本的解法是不要在 Reduce 里攒全量数据——如果你的单个 key 对应几百万条记录，说明数据模型本身就该拆分，硬扛不是办法。

Map 端的溢写行为也值得了解。Map 输出先进环形缓冲区，默认 100MB，写到 80% 就开始溢写，溢写之前按 key 排序。ReduceJoin 的 Map 输出量通常比输入还大（因为每条记录都被包装成了 Bean），所以适当调大mapreduce.task.io.sort.mb到 200 或者 300，能减少溢写次数，进而减少磁盘 IO 和后续的归并开销。我实测在一个千万行级别的模拟数据集上，把这个值从 100 调到 256，整体耗时降了差不多一成，不夸张但白拿。

5. 排查实录：ReduceJoin最常踩的六个坑

5.1 结果里只有订单没有产品名

这是最高频的问题，没有之一。现象是输出文件里订单号、数量都对，产品名一列全是空。原因几乎总是下面几种之一：key 的类型或格式不一致（前导零、大小写、空格）；文件名判断写反了（把订单当产品处理）；产品表的字段位置和代码里取值对不上。排查顺序我一般是：先cat输出目录里的part-r-00000看前几十行确认现象，再回看 Mapper 里 key 的赋值语句，最后打印几条原始数据比对。最快的验证办法是手动往两份小样本里塞能对上的 key，跑一遍看能不能拼出来，能拼通说明逻辑没问题，是数据的问题。

5.2 输出里出现"null"字符串

Java 里 null 拼接到字符串会变成 "null"。如果在 Reducer 里对没赋过值的字段直接toString，输出文件里就会看到刺眼的null。更麻烦的是writeUTF(null)会抛 NPE，所以序列化那一步就得拦。我的习惯是在 Bean 的 setter 里做兜底，或者在 write 里统一判空，双保险。还有一种情况是产品表里某行只有一列，fields[1]越界被 catch 掉了，这条记录压根没进 Reduce，最终结果里那个订单的产品名自然就空着——这种"静默丢弃"最坑，我的做法是 Mapper 里加个计数器，统计跳过的行数，跑完看计数器就知道丢了多少。

5.3 中文乱码与分隔符错乱

输出文件里的中文变成问号或者方块，八成是编码问题。Text.toString()默认按 UTF-8 解码，如果你的源文件是 GBK 存的，就会乱。解决办法是把源文件转成 UTF-8 再上传，或者明确指定输入编码。分隔符方面，除了前面说的逗号冲突，还有一个隐蔽的坑：从 Excel 复制出来的数据，分隔符可能是空格而不是 Tab，肉眼看着"对齐了"，实际是一个空格，split("\t")切不开，整行变成一个字段，处处越界。用cat -A看一眼行尾，制表符会显示成^I，一眼就能分辨。

5.4 Iterable里的对象全是最后一条

这个坑我在第 2.3 节提过，这里再强调一次具体症状：输出的每条订单，产品名都相同，而且恰好是产品表里最后一条的名字（或者订单数量都变成同一行）。原因就是 List 里存的是同一个被反复复用的对象引用。改法只有一个：拷贝。而且拷贝必须发生在context.write之前，不能拖到后面。有个更隐蔽的变体是：你在for循环里add(bean)，然后在循环外遍历 List 处理，觉得自己逻辑很清晰——结果一样是错的，因为对象还是那个对象。

5.5 任务跑到 Reduce 阶段失败

报错信息里出现java.lang.OutOfMemoryError: Java heap space或者Container is running beyond physical memory limits，说明 Reduce 端内存扛不住了。直接原因是单个 key 的 values 太多，全被塞进 List。处理方式分短长期：短期把 Reduce 内存调大、把 List 换成流式处理（减少内存占用但改动逻辑），长期还是得从数据分布上解决，比如拆分热点 key。还有一种容易漏的：Bean 里的字段用了String拼接不断累加，字符串对象在堆里堆成山，也是 OOM 的常见来源。

5.6 问题速查表

6. 从ReduceJoin到MapJoin：什么时候该换方案

6.1 DistributedCache广播小表

ReduceJoin 的短板很明确：两个表都要经历完整的 shuffle，网络传输和磁盘排序的开销都不小，而且只要有倾斜就会被单点拖住。如果其中一张表足够小（能完整塞进每个 Map 任务的内存），那就有更优解——在作业启动时把小表分发到每个 Map 节点，Map 阶段直接查内存里的映射表，关联在本地完成，Reduce 只负责输出。Hadoop 提供的机制叫分布式缓存，现在通过job.addCacheFile()使用：

job.addCacheFile(new URI("/input/product.txt#product")); // Mapper 的 setup 中读取 @Override protected void setup(Context context) throws IOException { BufferedReader br = new BufferedReader( new InputStreamReader(new FileInputStream("product"), "UTF-8")); String line; while ((line = br.readLine()) != null) { String[] f = line.split("\t", -1); if (f.length >= 2) { productMap.put(f[0], f[1]); } } br.close(); }

URI 后面的#product是给本地化文件起别名，读的时候直接用"product"这个相对路径，不用去拼什么绝对路径——不同节点上的缓存目录不一样，硬拼必错。mapper 里就简单了：读到一条订单，拿 pid 去productMap里get一下，拼好直接context.write。setNumReduceTasks(0)连 Reduce 都不用起，任务启动快得离谱。

不过这里也有坑。小表多大算"小"？我的判断标准是解压后不超过 200MB，因为每个 Map 任务都要在堆里放一份，Map 并发高的时候内存压力会成倍放大。另外 MapJoin 的输出文件数等于 Map 任务数，如果输入分片多，输出会碎成一地，需要额外合并。还有一点：小表更新了必须重新提交作业，缓存不会自动刷新，这在需要定期跑的场景里得靠调度器配合。

6.2 两种方案实测对比

我在一个模拟项目里做过一轮对照，数据是订单表约 2000 万行、产品表 3 万行，3 个节点的小集群，Reduce 个数统一给 4。这个数据量级不算大，但足够看出趋势。

耗时差距主要来自两处：一是 ReduceJoin 的 shuffle 和排序，二是 MapJoin 省掉了 Reduce 阶段的启动和归并。需要注意的是这个数字跟集群配置、数据分布强相关，换台机器就可能差出另一个结果，所以别把具体秒数当结论，看的是量级差异——在这个规模下差三倍，是值得换方案的理由。

6.3 我的选择原则

踩过几轮之后，我形成了一个比较粗的判断顺序。先看能不能不进 MapReduce：如果两张表都能放进一个普通数据库，直接用 SQL 关联，别为了用而用。确定要走批处理了，再看维表大小：维表能进内存就 MapJoin，这是绝大多数业务场景的实际情况，因为维表通常是产品、地区、类目这类相对稳定的数据。只有当两张表都很大、或者维表本身在持续膨胀的时候，才回到 ReduceJoin，并且要提前做好倾斜预案。

还有一个折中方案有人会问：能不能在 ReduceJoin 里做半连接，先在 Map 端过滤掉维表里没有的 key？可以，这是 MapReduce 优化里的一个经典技巧。加一个前置作业把维表的 key 抽出来做成小文件，广播到主作业的 Map 端做过滤，能砍掉大量无效记录，shuffle 数据量立刻下来。代价是多一个作业、多一轮调度。我用过一次，在一个维表覆盖率只有 30% 的场景里效果很明显，但如果覆盖率有 90%，那就纯属折腾。

最后分享一个我在调试这个案例时养成的小习惯：先用手工构造的五行数据跑通全流程，再上真实数据。五行数据里刻意放一个"订单有、产品没有"的 pid，放一个"产品有、订单没有"的，再放一个两边都有的。跑完看输出，如果那个"订单有产品没有"的 pid 对应的订单，产品名列是空的而不是整行消失，说明你写的是左连接语义；如果整行都没了，说明逻辑变成了内连接。这个小测试能一次性验出字段映射、空值处理、连接语义三类问题，比拿几百万行数据去跑、然后对着输出文件猜要高效得多。ReduceJoin 这个案例本身不复杂，真正花时间的从来都是这些边角上的确认。