凌晨两点，一个跑了一整晚的离线作业卡在 reduce 67% 不动，日志里每隔十分钟刷一次拉取失败，这种场面我在做数据平台那几年遇过不止一次。每次往上追，最后都绕回同一个东西——MapReduce 的执行模型。不少人对它的印象停留在“老古董”“被内存计算淘汰了”，但只要你还在碰离线批处理、还在给上层 SQL 引擎做底层调优、还要处理几十 TB 的落地日志，MapReduce 那套“切片、排序、归并、落盘”的路子就躲不开。它不只是一个框架，更是一种把大问题拆成可并行小问题的思考方式。

这篇东西我想把 MapReduce 从里到外拆一遍：它在解决什么问题，几个阶段各自干了什么，shuffle 为什么是性能命门，怎么写一个真能跑起来的作业，以及我踩过的那些坑。适合刚入门、能看懂 Java 但没自己写过作业的同学，也适合已经会用、但调优时心里没底的老哥。原理讲透，代码给全，参数给到具体数量级。

1. 先把问题摆出来：为什么单机一定扛不住

1.1 一个 20TB 日志的尴尬现场

假设你手上有一批按天滚动的服务端日志，一天 20TB，你要统计每个接口路径的调用次数、平均耗时、错误码分布。用最朴素的写法：读文件、逐行解析、往一个哈希表里累加，最后输出。单机跑，瓶颈会依次撞上三堵墙。

第一堵是磁盘 IO。机械盘顺序读算 150MB/s，20TB 光读完就要 20×1024×1024/150 ≈ 14 万秒，接近 39 小时，这还是理想顺序读。第二堵是内存。去重后的接口路径轻松上千万条，每条 key 加计数器按 200 字节估，就是 2GB 常驻，加上 JVM 对象头开销和 GC 压力，很容易在几个小时后 OOM。第三堵最要命：任何一步出错都得从头再来，中间没有可恢复的检查点，跑 30 小时崩在第 29 小时，等于白干。

分布式要解决的就是这三件事：把 IO 压力摊到多台机器、把内存压力切成多份、把失败成本降到只重算一小块。听起来简单，难的是怎么切、切完之后怎么合、合的过程中怎么保证不重不漏。

1.2 分而治之不是口号，是三次实实在在的切分

很多人说 MapReduce 是分治思想，这话没错但太虚。落到工程上，它其实做了三次不同性质的切分，这三次切分的对象和目的完全不同。

第一次切分针对输入数据，切出来的单位叫 split。一个 512MB 的文件按 128MB 切成 4 片，每片交给一个 map 任务。这一步只记录“从第几个字节读到第几个字节”，不复制数据，所以切分开销几乎为零。切片数量直接决定 map 任务的并行度，是并行度的源头。

第二次切分针对中间结果的键空间，切出来的单位叫分区（partition）。map 输出的是海量键值对，按 key 的哈希值把它们分成 N 份，每份最终交给一个 reduce。这一步决定了负载是否均衡，是数据倾斜的根源所在。

第三次切分针对单个分区内部的数据，切出来的单位叫归并段。reduce 拉取自己那份数据后，要先在内存里排序，内存不够就溢写到磁盘形成一个个有序小文件，最后做多路归并。这一步决定了单点内存压力和磁盘压力。

这三次切分对应三类典型故障：切片太少跑不满集群、分区不均导致长尾、归并段太多把磁盘打满。调优的时候先判断卡在哪一层，比盲调参数有效得多。

1.3 为什么核心链路坚持落盘而不是全程驻留内存

这是 MapReduce 最被诟病的设计，也是它最稳的地方。map 输出不直接推给 reduce，而是先写本地磁盘；reduce 拿到数据先落盘再归并。整个过程有大量磁盘往返，速度上确实吃亏。

但换个角度看：内存驻留方案一旦某个节点挂掉，丢失的是整条血缘链上的中间结果，重算代价可能是整个上游；而落盘方案里，挂掉只影响这个节点上正在跑的那几个任务，已经完成的 map 输出还在磁盘上，只需重跑失败的那一小块。在一个由几千台普通机器组成、磁盘和网卡每天都在随机出问题的集群里，这个取舍非常现实。

我的经验是：数据量在百 GB 级、迭代轮次多、对延迟敏感的活，用内存计算确实快；但数据量上到几十 TB、跑一次要几小时、中间还可能因为节点故障中断的场景，落盘带来的可恢复性比那点速度值钱。理解这个取舍，你就理解了 MapReduce 为什么长这样。

2. 五个阶段拆开看：数据在集群里到底走了哪几步

2.1 输入格式与切片：并行度从这里开始

作业启动时，客户端会先问输入格式要切片清单。默认实现按块大小切，同时做一个小优化：如果一片的剩余部分不超过块大小的 1.1 倍，就整片给前一个 split，避免出现极小的碎片片。这个 1.1 倍的阈值很多人不知道，但它会实实在在影响你的任务数——100 个 130MB 的文件，如果不做这个合并，会切出 100 个 128MB 加 100 个 2MB 的小片，多出 100 个几乎没有数据量的任务，光调度开销就够呛。

切片清单生成后，每个 split 会被分配到一个任务上。分配的优先级是数据本地性：最好分到数据所在节点（同机），其次同机架，最后才是跨机架。跨机架传输要走汇聚层交换机，带宽通常是机内的十分之一，所以调度器会尽量等本地资源，这个等待有超时，超时后仍然会跨节点起任务，不然小集群会饿死。

注意：切片是按字节切的，不是按行切的。一个 split 的边界很可能落在一行中间，读取时框架会跳过第一行的残片、并把最后一行读完，这部分逻辑由 LineRecordReader 处理。你自己实现输入格式时如果忘了处理边界，就会出现丢行或者重复行，而且极难复现。

2.2 Map 阶段：能做和不该做的事

map 函数的职责单一：拿到一行输入，吐出零到多个键值对。它的输入是(偏移量, 一行文本)，输出是(任意可序列化的 key, 任意可序列化的 value)。

有几件事我建议在 map 里做：数据清洗和脏行过滤，一条格式错乱的行直接 return，比让它流到下游触发异常划算得多；字段裁剪，只保留后续要用的列，中间结果能小一半；轻量的维度关联，把小的维度表通过分布式缓存加载到内存，在 map 端就完成 join，能省掉整个 shuffle。

有几件事千万别在 map 里做：初始化连接外部数据库或远程服务。map 任务可能成千上万，每个任务建一次连接，数据库连接池瞬间打满。正确做法是放到 setup 方法里并做懒加载，或者更稳妥地放到每个 JVM 的静态块里，因为一个 JVM 会复用跑多个任务。还有一件事是往 map 里塞超大对象，JVM 堆是固定分配的，塞爆了直接 OOM，而 OOM 的堆栈信息往往指向别处，排查起来很费劲。

map 输出的键值对不会立刻落盘，而是先写进一个环形内存缓冲区，默认 100MB。当缓冲区用到 80% 时，一个后台线程开始把数据排序后溢写到磁盘。这个比例调高能减少溢写次数，但留给自己处理的空间就小了，容易触发阻塞等待。

2.3 Shuffle：整个作业最贵的 70%

如果只让我讲一个 MapReduce 的知识点，我讲 shuffle。一个典型作业里，shuffle 阶段消耗的时间经常占到总时长的 60% 到 70%，剩下才是真正的计算。

shuffle 分两半。map 端这一半，缓冲区里的数据在溢写前会按 key 排序，排序用的是分区号加 key 的复合比较器：先按分区号排，保证同一个分区的数据在文件里连续；同分区内再按 key 排。这样 reduce 拉取时，一个分区就是文件里一段连续的区间，可以用一次范围读拿全。溢写文件如果有多个，会在 map 结束前做一次多路归并，合成一个带索引的大文件——索引记录的是每个分区在文件里的偏移量，reduce 靠这个索引精确读取自己那份。

reduce 端这一半，每个 reduce 任务会起若干个拉取线程，从所有已完成的 map 节点下载自己负责的那段数据。默认 5 个线程，这个值对中小集群够用，集群节点上千之后一定要往上调，否则拉取会成为瓶颈，你会在日志里看到大量“等待 map 输出”的提示。拉下来的数据先放内存，放不下就溢写，最后统一做归并排序，把相同 key 聚在一起喂给 reduce 函数。

这里有个容易被忽略的细节：reduce 必须等所有 map 都完成后才能进入最终归并，但拉取可以提前开始。只要某个 map 完成了，reduce 就会去拉它的输出，这叫 shuffle 的流水化。所以你会看到一个现象：map 完成 95% 的时候 reduce 已经拉到 90% 的数据了，最后卡在那 5% 的慢 map 上，这段时间叫“长尾”。推测执行就是为这个设计的。

2.4 Reduce 与输出：归并之后才是业务逻辑

reduce 函数接收的是(key, 该 key 的所有 value 的迭代器)。注意是迭代器不是集合，框架为了省内存，通常不会把这个 key 的所有 value 一次性加载到内存，而是边归并边喂。这意味着两件事：第一，你不能对 values 做两次遍历，第二次会拿到空；第二，你不能假设 values 的顺序，如果业务依赖顺序，得自己在 value 里带上排序字段处理。

输出格式负责把结果写到目标存储。默认是写到文件系统的文本文件，每个 reduce 任务产生一个文件。写 20 个 reduce 就会产生 20 个part-r-00000到part-r-00019的文件，这也是为什么小文件问题在 MapReduce 里特别突出——下游如果再读这些文件，输入切片又会碎一次，恶性循环。

最后说下计数器。框架内置了一堆指标：读取的记录数、写入的记录数、溢写次数、拉取的字节数、失败的拉取次数。这些数字看着不起眼，排查问题时是黄金线索。溢写次数异常多说明缓冲区太小或者数据分布有问题；失败拉取次数高说明网络或者节点有问题；map 输入记录数和输出记录数差距巨大，通常是过滤逻辑写错了。

3. 自己动手写一个能跑的作业

3.1 环境与依赖

本地验证用单机伪分布式最省事，装好 JDK 8 以上和 Maven，引入三个依赖就够：common、client、mapreduce-client-core，版本保持和集群一致，版本不一致是最常见的“本地能跑集群报错”原因。

<dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>3.3.6</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-mapreduce-client-core</artifactId> <version>3.3.6</version> </dependency>

打包时注意把依赖的 scope 设成 provided，因为集群上已经有这些 jar，打进去会让作业包从几 MB 膨胀到上百 MB，每次上传都慢，还可能出现类冲突。用 Maven 的 shade 插件打一个瘦包，只带业务类和第三方小依赖，是最稳妥的做法。

3.2 Mapper 与 Reducer 的骨架

拿一个按接口路径统计调用量的例子，完整代码长这样：

public class ApiCallCounter { public static class TokenizerMapper extends Mapper<LongWritable, Text, Text, IntWritable> { private static final IntWritable ONE = new IntWritable(1); private final Text outKey = new Text(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); String[] parts = line.split("\t"); // 字段不够直接丢弃，别让一条脏数据拖垮整个 split if (parts.length < 3) { context.getCounter("custom", "bad_line").increment(1); return; } String path = parts[1]; if (path.isEmpty()) { context.getCounter("custom", "empty_path").increment(1); return; } outKey.set(path); context.write(outKey, ONE); } } public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> { private final IntWritable result = new IntWritable(); @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable v : values) { sum += v.get(); } result.set(sum); context.write(key, result); } } }

这段代码有两个细节值得说。一是复用对象：outKey和ONE定义成复用实例而不是每次 new，因为 map 可能被调用上亿次，少创建上亿个短命对象，GC 压力能明显降低。这是 MapReduce 编程里最值钱的一个小习惯。二是自定义计数器：脏数据不是简单丢掉就完事，要能看见丢了多少，不然数据对不上时你会怀疑人生。

3.3 Combiner、Partitioner 和驱动配置

Combiner 是 map 端的预聚合，本质上就是在本地的 map 输出上先跑一遍 reduce 逻辑。上面这个求和场景，一个 map 输出 100 万条(path, 1)，加了 combiner 之后可能只剩 5 万条(path, 累计值)，网络传输量直接降一个数量级。

job.setCombinerClass(IntSumReducer.class);

但 combiner 不是随便加的。它要求操作满足结合律和交换律，求和、求最大值、求并集可以，求平均值不行——你会得到各分片平均值的平均，而不是全局平均。这个坑我见过不止一个团队踩。真要算平均值，就把 sum 和 count 一起输出，最后在 reduce 里除。

分区器决定 key 去哪个 reduce。默认用 key 的哈希值取模，对绝大多数场景够用。但当你要保证同一类 key 落到同一个 reduce、或者要做二次排序时，就得自己写：

public class PathPrefixPartitioner extends Partitioner<Text, IntWritable> { @Override public int getPartition(Text key, IntWritable value, int numPartitions) { String path = key.toString(); // 按一级路径聚合，保证 /api/user/* 全落到同一个 reduce int idx = path.indexOf('/', 1); String prefix = idx > 0 ? path.substring(0, idx) : path; return (prefix.hashCode() & Integer.MAX_VALUE) % numPartitions; } }

注意& Integer.MAX_VALUE这一步不能省。哈希值可能是负数，负数取模结果还是负数，会直接抛数组越界，而且只在特定 key 上偶发，测试时未必能发现。

驱动部分把参数串起来：

Configuration conf = new Configuration(); conf.set("mapreduce.map.memory.mb", "3072"); conf.set("mapreduce.map.java.opts", "-Xmx2457m"); conf.set("mapreduce.reduce.memory.mb", "6144"); conf.set("mapreduce.reduce.java.opts", "-Xmx4915m"); conf.set("mapreduce.task.io.sort.mb", "300"); Job job = Job.getInstance(conf, "api-call-counter"); job.setJarByClass(ApiCallCounter.class); job.setMapperClass(TokenizerMapper.class); job.setCombinerClass(IntSumReducer.class); job.setPartitionerClass(PathPrefixPartitioner.class); job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); job.setNumReduceTasks(30); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); // 输出目录必须不存在，否则直接抛异常，这是新手最常见的第一个报错 System.exit(job.waitForCompletion(true) ? 0 : 1);

内存参数有个固定的换算关系：JVM 堆大小设为容器内存的 0.75 到 0.8。上面 3072MB 配 2457m 就是这个比例。堆设得和容器一样大，进程会因为堆外内存被系统杀掉，日志里通常只有一个含糊的“被信号杀死”，查半天。

3.4 reduce 数量怎么估，参数怎么配

reduce 数量是调优里最常被问的问题。我的估算方法分三步。

第一步，算单 reduce 处理的数据量。中间结果总量除以 reduce 数，控制在 1GB 到 2GB 之间比较舒服。太少则 reduce 任务多、调度开销大、输出小文件多；太多则单个 reduce 跑得久，一旦失败重算代价高。

第二步，算中间结果总量。原始数据 10TB，map 端做了字段裁剪和过滤后可能只剩 3TB，加了 combiner 之后可能压到 300GB。按 1.5GB 一个 reduce 算，就是 200 个。

第三步，看集群资源上限。每个 reduce 占 6GB 内存，集群总共可用的 reduce 槽位是 500 个，那 200 个完全跑得开，可以再调大到 300 缩短单任务时长。如果算出来 2000 个，槽位只有 300，就得砍到 300 左右，多跑几轮。

常见参数的参考值我整理成一张表，都是实测比较稳的量级：

举个具体的溢写次数估算：某个 map 输出 2GB 数据，缓冲区 300MB，溢写阈值 0.8，那么每次溢写 240MB，需要溢写 2×1024/240 ≈ 9 次，最后再归并这 9 个文件。如果把缓冲区调到 100MB（默认），就是 25 次溢写加 25 路归并，磁盘压力和耗时差出一截。这就是为什么我通常建议把io.sort.mb直接提到 300 左右。

4. 排查实录：那些让作业卡住的真实原因

4.1 数据倾斜：reduce 卡在 99% 的经典场景

症状很典型：30 个 reduce，29 个在半小时内跑完，剩下一个卡在 99% 加了两个小时还没动。打开这个 reduce 的日志，会发现它在反复溢写，说明单个 key 的数据量远超预期。

根因通常是某个维度的值极度集中。比如按用户 ID 聚合，有几个爬虫账号贡献了 40% 的记录；或者 join 时某个默认值（空字符串、-1、未知）占了绝大多数，它们全都哈希到同一个分区。

我的处理顺序是这样的。先看能不能在 map 端消掉：如果是大小表 join，把小的那张表通过分布式缓存加载进内存，map 端直接完成关联，整个 shuffle 都省了，这是收益最高的一招。如果两边都大，就看能不能加 combiner 预聚合，把重复 key 先在本地合掉。

前两招都不行，就上打散。给 key 加随机后缀，比如key_0到key_9，把原来一个 reduce 的活分给十个；reduce 端拿到带后缀的 key 先做一次局部聚合，再加一个只做汇总的第二个作业。两阶段聚合听起来麻烦，但处理严重倾斜时是唯一稳的办法。

还有一种容易被忽略的倾斜来源：分区器写错了。我见过一个自定义分区器直接用key.hashCode() % numPartitions，忘了处理负数，结果一部分数据全跑到负索引上，异常被吞掉后又重试，表现就是某个 reduce 迟迟不结束。这类问题看计数器里的失败任务数能发现端倪。

4.2 内存问题：OOM 和 GC 是两条不同的线索

OOM 分两种，处理方式完全不同。

一种是堆内存溢出，日志里明确写着java.lang.OutOfMemoryError: Java heap space，堆栈指向你的业务代码。这多半是在 map 或 reduce 里缓存了不该缓存的东西，比如把 values 转成 ArrayList 再遍历、或者初始化了一个巨大的本地 Map。解决办法是改逻辑，实在需要就调大堆，但调大之前先想清楚数据量会不会继续涨。

另一种是被系统杀掉，日志里没有 Java 异常，只有一个容器退出码 137，或者一句“Container killed by the system”。这通常是物理内存超了，堆外内存（直接缓冲区、JVM 元空间、线程栈）没算进去。这时候要做的是压低堆大小，而不是调大，很多人反着调，越调越容易被杀。

还有一种是 GC 时间过长被杀，退出信息里会带“超出 GC 时间上限”。这一般是堆设得偏大但对象存活率高，或者堆偏小导致频繁 Full GC。判断方法是看 GC 日志里 Full GC 后的老年代占用，如果回收后仍然接近满，就是真的放不下，得改逻辑；如果回收后降下来很多，就是堆太小。

4.3 小文件：并行度不是越大越好

新手的直觉是任务越多越快，实际不是。假设输入是 10 万个 10MB 的小文件，默认切片策略会切出 10 万个 map 任务。每个任务的启动开销（JVM 启动、配置加载、连接建立）大概 1 到 3 秒，10 万个就是 30 到 80 小时纯启动开销，还没算上调度队列的排队和 namenode 的元数据压力。

解决办法有两层。第一层是入口合并，用合并输入格式把多个小文件塞进一个切片，按 128MB 的目标大小攒够再切，任务数能直接降到千级。第二层是出口控制，reduce 数量别拍脑袋定大，如果最终结果只有 500MB，用 100 个 reduce 就是 100 个小文件，下游又得处理一遍，用 5 到 10 个更合适。

提示：判断是否存在小文件问题，看两个计数器——启动的 map 任务数和平均每个任务的输入字节数。后者如果远小于块大小（比如只有几 MB），基本可以确认是切片过碎。

4.4 常见问题速查表

最后一行那个坑值得单独说一句。如果同一个任务在多个不同节点上反复失败，那就不是硬件问题，是代码里存在非确定性——比如依赖了系统时间、随机数、或者 HashMap 的遍历顺序。这类 bug 在单机测试时完全看不出来，分布式一跑就现原形，而且日志里往往只有一句“任务失败次数超过上限”，非常有迷惑性。

5. 这套模型今天还能给你什么

5.1 它在现代数据栈里的真实位置

直接写 MapReduce 作业的场景确实少了，但它的模型没有消失，而是下沉成了基础设施。上层那些用 SQL 描述的计算，最终编译执行时，聚合和 join 的物理算子仍然是“先局部聚合再全局归并”这一套；分布式文件系统里的数据布局、分区表的目录结构、甚至列式存储里的行组划分，本质都是切片的延伸。

我个人的判断是：不需要再手写大量 MapReduce 作业，但必须懂它的执行模型。因为当你调优一个上层查询时，你调的每一个参数背后对应的都是这套模型里的某个环节——并行度对应切片，倾斜对应分区，内存溢写对应缓冲区。不懂模型，调参就是碰运气。

5.2 三个能直接迁移到别处的设计习惯

第一个习惯是先想清楚“怎么切”，再想“怎么算”。不管是处理日志、批量生成报表、还是做图片批处理，先确定切分维度和每片的处理成本，比直接写循环要重要得多。切分维度选错了，后面再怎么优化都是在错误的方向上努力。

第二个习惯是把“局部预聚合”当成默认动作。能在数据产生的地方先合一次，就别让它原样流到下一层。这个思路在缓存设计、接口聚合、日志上报里全都适用，本质就是 combiner 的思想：减少传输量永远比提升传输速度划算。

第三个习惯是让每个环节都可重入。MapReduce 的任务失败能重跑，是因为每个任务不依赖外部状态，重跑结果一致。写批处理脚本、写数据同步逻辑时，尽量保证“跑一次”和“跑三次”结果相同，这个约束会逼着你把状态管理做干净，长期看省下的排查时间远超当初多花的设计时间。

真要说这套东西给我留下的最深印象，是它教会我用“失败是常态”的视角做设计。单机思维里失败是意外，分布式思维里失败是日常，节点会挂、磁盘会坏、网络会抖，把这些当成前提而不是异常，设计出来的东西才经得起跑。

调优这事儿没有银弹。我现在的习惯是每次作业跑完把计数器拉出来看一眼：map 输入输出比、溢写次数、拉取字节数、失败任务数，这四个数字基本能告诉你这次跑得健不健康，以及下次该往哪个方向调。数据量涨十倍的时候，这套观察方法比背参数表有用得多。