企业官网建设流程全解析

1. 先想清楚：MapReduce 这个计算框架到底卡在什么位置上

聊 MapReduce 之前，我习惯先把它的位置摆正。它不是一个数据库，不是一个存储系统，也不是一个查询引擎，它是一套把"一个大计算"拆成"很多个小计算"再合并结果的编程模型，外加一个负责调度、容错、数据搬运的执行框架。名字里的 Map 和 Reduce 借的是函数式编程里的两个老概念：Map 做逐条转换，Reduce 做归并聚合。真正值钱的部分不在概念本身，而在于它把分布式执行里最脏最累的活——任务切分、节点调度、失败重试、中间数据搬运、结果排序——全部收编进了框架，写业务的人只需要填两个函数。

这套东西解决的核心问题是：单机内存装不下、单机 CPU 算不完的批量数据处理。比如几百 GB 的日志要做分组统计、TB 级的文本要做倒排索引、全量订单表要按天做多维聚合。这类任务的特点是数据量大、计算逻辑相对简单、对延迟不敏感、结果要求准确。只要你的场景落在这个框里，MapReduce 就是一台皮实耐用的老卡车，慢，但你不用担心它半路撂挑子。

我把它的适用边界总结成四个假设，理解了这四条，基本就不会用错：

• 数据是批量的、有边界的。它的输入是"已经躺在存储里的文件集合"，不是一个持续到达的流。流式场景硬套批处理，只能靠不断切小批次，反而把框架的调度开销放大了。
• 计算可以按输入分片独立进行。每个分片算完自己的部分，不需要知道别的分片算到哪了。一旦你的逻辑需要跨分片频繁通信，比如迭代式的图计算、机器学习训练，框架就变成了累赘。
• 中间结果落盘是可以接受的代价。这是它和内存计算框架最本质的区别，也是它慢的根源。落盘换来的好处是：任何一个环节挂掉，都能从最近的落盘点重来，不需要从头重算整条链路。
• 延迟以分钟到小时计。一个作业的启动开销、调度开销、Shuffle 开销加起来，通常就是几十秒起步。你想拿到亚秒级响应，那是另一类系统该干的事。

注意：很多人一上来就问"MapReduce 是不是过时了"。这个问题问错了方向。批处理的思想没有过时，过时的是"所有批处理都该用它"这种用法。今天它更多作为离线数仓的地基存在，上面跑的是更高层的 SQL 引擎，但底下的分片、Shuffle、容错逻辑换汤不换药。

适合读这篇内容的人分三类：一是在离线计算平台上排查过"作业卡在 reduce 99%"的人；二是要写 MapReduce 作业但没系统搞懂 Shuffle 的人；三是想理解分布式计算框架设计取舍的人。第三类其实收益最大，因为 Shuffle 这一课，在任何分布式系统里都躲不掉。

2. 编程模型拆解：Map 和 Reduce 之间发生了什么

2.1 从词频统计看清模型骨架

词频统计是这门课的"Hello World"，但大部分人只记住了代码，没记住它暴露出的模型约束。用最朴素的写法，输入是一堆文本文件，Map 阶段逐行读，切词，每遇到一个词就吐一个<词, 1>；框架把所有 Map 的输出按 key 分组，把同一个词的所有 1 聚到一起交给 Reduce；Reduce 把这一串 1 加起来，输出<词, 总数>。

这段流程里藏着三个关键约定，写业务代码时如果违反，框架不会报错，但结果一定是错的：

第一，Map 和 Reduce 都必须是纯函数式的。给定相同的输入，必须产出相同的输出，不能依赖外部可变状态，不能有随机性，不能依赖执行顺序。原因很直白：框架可能在任意节点、任意时刻重跑某个任务，重跑的结果必须和第一次一致。我见过有人在 Map 里写System.currentTimeMillis()打时间戳，作业一旦发生重试，同一批数据就出现了两个不同时间戳，下游去重直接失效。

第二，数据流只能单向。Map 的输出只能流向 Reduce，Reduce 之间不能互相通信，也不能跨 key 传递信息。这条限制是框架能大规模并行的前提。真需要跨 key 的信息，只能拆成两个作业串起来，前一个作业的输出作为后一个的输入。

第三，分组是按 key 的完整值做的。框架不关心你的 key 是什么类型，它只做一件事：把所有相同 key 的记录聚到一起。所以 key 的设计直接决定了数据分布，也直接决定了会不会倾斜。

下面这段是简化后的 Java API 风格骨架，只看结构，不看包名，各发行版的类路径略有差异：

// Map：输入一行文本，输出若干 <词, 1> public class TokenMapper extends Mapper<LongWritable, Text, Text, IntWritable> { private final Text word = new Text(); private final IntWritable one = new IntWritable(1); @Override protected void map(LongWritable offset, Text line, Context ctx) throws Exception { // 按非字母字符切分，注意不要用 split("\\s+") 处理含标点的中文混排 String[] tokens = line.toString().toLowerCase().split("[^\\p{L}\\p{N}]+"); for (String t : tokens) { if (t.isEmpty()) continue; word.set(t); ctx.write(word, one); } } } // Reduce：把同一个词的计数累加 public class SumReducer extends Reducer<Text, IntWritable, Text, IntWritable> { private final IntWritable result = new IntWritable(); @Override protected void reduce(Text key, Iterable<IntWritable> values, Context ctx) throws Exception { int sum = 0; for (IntWritable v : values) { sum += v.get(); } result.set(sum); ctx.write(key, result); } }

驱动部分需要显式指定 Mapper、Reducer、输出 key/value 类型、输入输出路径。这里有个新手高频踩的坑：输出 key/value 类型必须和 Reducer 声明的泛型完全一致，否则作业能提交成功，跑到一半才抛类型转换异常，日志里还埋得很深。我一般的习惯是先把类型写成最宽的形式调试通，再收窄。

2.2 分区函数决定了数据往哪走

Map 输出<k, v>之后，框架要做一件关键的事：决定这条记录交给哪个 Reduce 处理。这个决定权在Partitioner手里，默认实现是hash(key) % reduceTaskNum。

这个默认实现看似公平，实则是倾斜的罪魁祸首之一。哈希只保证大体均匀，一旦某个 key 的数据量本身就是其他 key 的几十倍，比如日志里某个异常码占了 40% 的记录，那么无论怎么哈希，它都会整个压到同一个 Reduce 上。这时候你看到的现场是：99 个 Reduce 在十分钟内跑完，最后 1 个卡在 99% 跑了两个小时。

解决办法后面第 4 节会细说，这里先记住一条原则：分区函数的输出必须是确定性的，并且分区数量要和 Reduce 数量严格对应。自定义 Partitioner 时，返回值必须落在[0, numReduceTasks)区间内，越界会直接抛异常；同时不要在分区函数里做任何依赖外部配置的随机化，否则重试时数据会落到不同分区，结果就散了。

2.3 Combiner：最容易被忽略也最容易用错的一层

Combiner 在概念上是"Map 端的迷你 Reduce"，它在本节点内先做一次局部聚合，把<a,1> <a,1> <a,1>压成<a,3>再发出去。它的价值在网络传输上：如果每个词平均出现 100 次，加上 Combiner 之后 Shuffle 的数据量能降到原来的百分之一，这个收益相当可观。

但 Combiner 有一条铁律：它只能用于满足结合律和交换律的运算。求和、求最大值、求并集，这些都可以。求平均值不行——avg(avg(1,3), 5)等于 3，而真实的avg(1,3,5)也是 3，看着对，但换个数字就错：avg(avg(1,3), 100)是 51，真实值是 34.7。想让平均值也能用 Combiner，必须把中间结果改造成(sum, count)这样的复合结构，在 Reduce 里再做除法。

还有一点很多人不知道：框架不保证 Combiner 一定被调用。它可能调用 0 次、1 次或多次，完全取决于溢出文件的合并情况。所以 Combiner 的逻辑必须和"不调用"时的结果完全等价，任何有副作用的操作（写文件、发消息、改全局计数器以外的状态）都不要放进去。

提示：判断一个聚合能不能加 Combiner，最快的自测方法是问自己——把输入任意切分成两段分别算，再把两个结果合并，答案和一次算完是否相同。相同就能加，不同就得改结构。

3. 执行引擎的运行时细节：一个作业到底怎么跑完的

3.1 从提交到输出：五个阶段的完整链路

作业提交之后，框架做的事情比大多数人想象的多。我把这条链路拆成五段，每一段都对应一个可能出问题的地方。

第一段，输入分片。输入格式化组件扫描输入目录下的所有文件，按分片规则切成若干逻辑分片，每个分片生成一个 Map 任务。分片大小默认等于存储块大小（常见是 128MB），计算规则是max(minSize, min(maxSize, blockSize))。这里要注意：分片是逻辑切分，不是物理切分，一个分片可能跨两个块，框架会自动处理跨块读取。真正影响任务数的是分片大小，不是文件个数——1000 个 1MB 的小文件会生成 1000 个 Map 任务，每个任务的处理时间还没调度开销长，这就是著名的小文件问题。

第二段，Map 执行。每个 Map 任务读取自己的分片，逐条调用 map 函数，输出写进一个环形缓冲区。这个缓冲区默认 100MB，填充到 80% 时触发一次溢出（spill）。溢出前，框架会先把缓冲区里的记录按"分区号 + key"排好序，如果配了 Combiner 就先局部聚合，然后写到本地磁盘的临时文件。一个 Map 任务可能溢出多次，产生多个临时文件。

第三段，Map 端归并。Map 任务结束前，把所有临时文件做一次多路归并，合并成一个大文件，归并的同时仍然保持分区内有序。这个有序性是后面 Reduce 能高效归并的基础，也是很多人以为"MapReduce 会自动全局排序"的误解来源——Map 端只保证分区内有序，全局有序需要额外配置或者二次作业。

第四段，Shuffle 拉取。Reduce 任务启动后，通过心跳向各 Map 节点请求属于自己的那部分数据，默认并行拉取 5 个。拉回来的数据先放内存缓冲，缓冲满了写磁盘，边拉边归并。这一段是整个作业里网络和磁盘压力最大的地方，也是最容易出问题的地方。

第五段，Reduce 执行与输出。所有数据拉齐并归并成一个有序流后，框架按 key 分组，把同一个 key 的所有 value 作为一个迭代器传给 reduce 函数，输出直接写到目标路径。输出目录必须不存在，否则作业在提交阶段就会拒绝执行，这是框架的防呆设计。

3.2 Shuffle 细节与缓冲区参数怎么算

Shuffle 是这门技术里最值得深挖的一段。它的本质是一次分布式的排序 + 分发，涉及三个可调的关键量：缓冲区多大、什么时候溢出、归并多少路。

先看缓冲区。环形缓冲区大小由参数控制（多数发行版默认 100MB），溢出阈值默认 0.80。这意味着实际可用于排序的内存是 80MB。假设某个 Map 任务的输出总量是 200MB，那么它大约会溢出 2 次（160MB 触发两次，剩 40MB 在任务结束时直接参与最终归并）。

这个计算很粗糙，但足够用来判断参数该不该调。判断逻辑是这样的：如果单个 Map 的输出量远大于缓冲区，溢出次数多，每次溢出都是一次排序加一次磁盘写，属于纯浪费。这时候把缓冲区从 100MB 提到 200MB 到 400MB，溢出次数能直接减半。但也不是越大越好，缓冲区是从 Map 任务的堆内存里划走的，堆一共就那么大，缓冲区吃多了，业务代码里的集合对象就容易 OOM。我的经验值是：堆内存 2GB 的容器，缓冲区给 200MB 到 300MB 比较稳。

再看归并路数。Map 端和 Reduce 端的归并因子默认都是 10，含义是一次最多同时打开 10 个文件做归并。如果溢出文件有 50 个，那就得分 5 轮归并，每轮产生一个新文件，磁盘 IO 翻倍。把归并因子提到 40 到 100，能显著减少归并轮次，代价是同时打开的文件句柄变多，需要确认系统的文件句柄上限够用。

Reduce 端的拉取并行度默认是 5，这个值在现代网络环境下偏保守。集群网络是万兆的话，提到 10 到 20 通常能缩短 Shuffle 阶段的时间。但也要看 Reduce 节点的磁盘能不能扛住并发写，盲目拉高有时候反而更慢。

我把几个常用参数整理成表，方便对照。不同发行版的参数名前缀可能略有差异，但语义是一致的：

3.3 容错：框架最值钱的那部分能力

MapReduce 真正难写的地方不是 map 和 reduce 两个函数，而是它怎么在几百台机器随时可能挂掉的情况下，保证作业最终能跑完且结果正确。这部分逻辑有三层。

第一层，任务级重试。任何一个 Map 或 Reduce 任务失败，框架会在另一个节点重新调度它，默认重试 4 次（可配）。Map 任务失败重跑很干净，因为它不依赖任何中间状态，输入分片还在，从头算一遍就行。Reduce 任务失败重跑则有个优化：已经拉取完成的 Map 输出不需要重新拉，只需要重新做归并和 reduce 计算。这个细节在排查"为什么 Reduce 重试比重跑还慢"的时候很有用——如果 Map 输出已经被清理，那 Reduce 重试就得重新触发上游重算。

第二层，推测执行。集群里总会有那么一两个节点，因为磁盘老化、网络抖动或者别的任务抢占资源，跑得特别慢。框架通过进度报告发现某个任务的进度显著落后于同批次任务的平均值，就会在另一个健康节点上启动一个备份任务，两个任务谁先完成谁算数，另一个被 kill 掉。这个机制默认开启，对长尾任务效果明显。但它也有代价：会额外占用集群资源，在资源本来就紧张的时候，可能把整体拖得更慢。如果是那种"结果必须确定、不能重复执行"的场景，还得注意备份任务的副作用问题——虽然框架保证只有一个任务的输出被采纳，但任务执行过程中的外部写操作是拦不住的。

第三层，数据本地性调度。理想情况下，Map 任务被调度到数据所在的节点上，直接读本地磁盘，这叫节点本地。次优是同一个机架内的另一个节点，最差是跨机架。这个优先级不是随便定的：跨机架读一次数据的网络开销，可能是本地读的几十倍。所以调度器会尽量等一等，看能不能等到本地资源空出来。这个"等"是有超时限制的，等太久就直接调度到远端节点，避免整个作业被饿死。

注意：数据本地性在容器化部署的集群里经常失效。如果计算节点和存储节点是两套独立的机器，那所有 Map 任务都是远端读取，本地性优化等于没开。这种情况下的性能瓶颈往往不在计算，而在网络带宽，调 Map 的并行度才有意义。

4. 参数计算与实操调优：一次真实的倾斜治理

4.1 Reduce 数量到底怎么定

Reduce 数量是个被讨论烂了但依然有人拍脑袋定的话题。定得太少，单个 Reduce 处理的数据量太大，容易 OOM 或者跑成超长尾；定得太多，每个 Reduce 分到的数据很少，但每个 Reduce 都要经历一遍完整的启动、拉取、归并流程，调度开销和文件句柄开销反而上去，而且输出会变成一堆小文件，给下游添麻烦。

我用的估算公式是：Reduce 数 ≈ 0.95 × 集群可用 Reduce 槽位总数。系数取 0.95 而不是 1.0，是为了留一点余量，让部分 Reduce 提前完成时，剩下的任务还能利用空出来的槽位，形成流水，避免最后一批 Reduce 同时启动、同时拉数据把网络打满。

举个具体例子。假设集群有 10 个计算节点，每个节点配置为 16 核、64GB 内存，Reduce 容器申请 4GB 内存，那么单节点理论可跑 16 个容器，但要给系统和 Map 任务留余量，实际按 8 个 Reduce 槽位算。总槽位 80，乘 0.95 约等于 76，向上取整到 80。如果输入数据是 1TB，那么平均每个 Reduce 处理约 13GB，这个量级在合理区间。

再反推单个 Reduce 的数据量是否安全：Reduce 端归并时需要把部分数据放在内存里，默认缓冲占堆内存的 70%，4GB 堆就是 2.8GB。13GB 的数据里，最后一段归并流的记录会常驻内存，如果单条记录比较大（比如包含长文本字段），2.8GB 缓冲可能撑不住，会频繁触发磁盘归并。这种情况要么把 Reduce 数加大，要么把单条记录瘦身，要么把容器内存提上去。三个方案里，我最先尝试的是记录瘦身——在 Map 端就把不需要的字段丢掉，往往比加资源更划算。

4.2 一个倾斜现场的排查与改造

说个我实际处理过的案例。一个按用户 ID 聚合行为数据的作业，输入约 800GB，Reduce 设了 100 个。前 30 分钟一切正常，80 多个 Reduce 陆续完成，剩下的十几个卡在 90% 以上不动，其中一个卡在 99.8% 整整两个小时。

排查顺序是这样的：先看计数器，发现那个卡住的 Reduce 读取的记录数是其他 Reduce 的三十多倍；再看输入数据分布，发现有个固定的默认值占了全量数据的 35% 左右，这些记录全部落到同一个 Reduce。这就是典型的key 倾斜。

第一反应是加 Combiner。加了之后 Shuffle 数据量确实降下来了，但倾斜没解决——因为 Combiner 只减少传输量，不改变"同一个 key 必须去同一个 Reduce"这个事实。35% 的数据压在一个 Reduce 上，传输量再小，计算量还是在那里。

第二反应是自定义 Partitioner。思路是把那个热点 key 打散到多个 Reduce 上。实现上给热点 key 拼一个随机后缀，比如default_0到default_9，分区函数按新的 key 哈希，这样它能分散到 10 个 Reduce 上。但这里有个必须处理的后续问题：打散之后，同一个逻辑 key 的结果被拆成了 10 份，需要再做一次聚合。所以整个方案变成两阶段作业：第一阶段打散聚合出部分结果，第二阶段把后缀去掉再聚一次。第二次的数据量已经很小了，几秒钟就能跑完。

第三反应，也是我最后采用的方案，是两阶段聚合 + 热点单独处理。具体做法是：Map 端先做一次本地预聚合，把<用户ID, 事件>压成<用户ID, 计数>，这一步能把数据量压掉一个数量级；然后在 Partitioner 里判断，如果 key 等于那个热点值，就按记录内容的哈希分散到后 20 个 Reduce，其余 key 按正常哈希走前 80 个 Reduce；最后跑一个轻量的合并作业收尾。

改造后的效果：整体作业时间从 3 小时 40 分降到 48 分钟，卡尾现象基本消失。这个案例给我的教训是，倾斜治理要在数据进入 Shuffle 之前动手，等到 Reduce 阶段才发现，代价已经付完了。

4.3 一段可直接跑的流式写法

如果不想引入编译依赖，流式接口是最省事的入口，用标准输入输出就能跑。Map 脚本从标准输入逐行读，输出key\tvalue；Reduce 脚本从标准输入读已按 key 排序的数据，遇到 key 变化就输出上一组的聚合结果。

# mapper.py import sys for line in sys.stdin: line = line.strip() if not line: continue for token in line.lower().split(): # 只保留字母数字，去掉标点 token = ''.join(ch for ch in token if ch.isalnum()) if token: print(f"{token}\t1")

# reducer.py import sys current_key = None current_sum = 0 for line in sys.stdin: line = line.rstrip("\n") if not line: continue key, value = line.split("\t", 1) if key == current_key: current_sum += int(value) else: if current_key is not None: print(f"{current_key}\t{current_sum}") current_key = key current_sum = int(value) if current_key is not None: print(f"{current_key}\t{current_sum}")

提交时把两个脚本作为参数传进去，指定输入输出路径和 Reduce 数量即可。这套写法的性能比原生 API 差一些，主要差在序列化和进程启动开销上，但开发效率高得多，特别适合做数据探查和一次性统计任务。真要上生产的大作业，我还是建议用原生写法，或者干脆用更高层的 SQL 引擎把作业生成出来。

提示：流式写法里有个隐坑——Reduce 端收到的数据虽然是按 key 排好序的，但排序是按字节序做的，不是按数值或拼音。如果你的 key 是数字字符串，"10" 会排在 "9" 前面。需要数值序就补零对齐，需要中文拼音序就自己加个排序前缀。

5. 常见问题与排查技巧实录

5.1 三个高频故障现场

现场一：作业卡在 map 0% reduce 0% 超过十分钟。这个现象九成不是计算问题，而是资源申请不到。要么是队列资源被占满，任务在排队；要么是申请的容器内存超过了单节点可分配上限，永远满足不了。查调度日志能看到具体的排队原因。还有一种可能是输入路径不存在或者无权限，但这种通常在提交阶段就报错了，轮不到卡住。

现场二：Reduce 阶段报内存溢出。报错信息里如果出现"归并时内存不足"或者"拉取缓冲超限"，方向就是 Reduce 端内存吃紧。处理顺序建议是：先看是不是数据倾斜导致单个 Reduce 数据量暴增，再看是不是单条记录太大，最后才是调容器内存。反过来做，容易陷入"加内存—还是 OOM—再加内存"的死循环。

现场三：作业成功但结果少了数据。这类问题最阴险，因为不报错。排查路径有三条：一看 Map 输入记录数和源文件的实际记录数是否一致，排除输入格式解析漏行；二看 Reduce 输出文件数是否等于设定的 Reduce 数，如果有 Reduce 输出为空文件是正常的，但如果某个分区完全没有输出，可能是分区函数写错导致数据越界被丢弃；三看是否有 Combiner 把不该合并的东西合并了，尤其是涉及去重和计数的逻辑，Combiner 写错会导致结果偏小。

我把这些整理成速查表：

5.2 几条不上文档的实操心得

第一条，先在样本数据上跑通全链路，再上全量。我吃过这个亏：逻辑写对了，但在 1GB 样本上跑得好好的作业，上了 800GB 全量之后因为缓冲区参数不合理，溢出次数暴增，跑了六个小时。后来养成习惯，样本跑通之后，先用 5% 的数据量做一次全链路压测，看 Shuffle 数据量、溢出次数、单任务耗时这些计数器指标，估出全量的耗时再提交。多花的这半小时，经常能省下几个小时的返工。

第二条，计数器比日志好用得多。框架内置了一堆计数器，Map 输入记录数、Map 输出记录数、Reduce 输入记录数、溢出的记录数、合并的记录数，这些数字能直接告诉你数据在哪一步发生了变化。我排查"结果少了"这类问题，第一步永远是拉计数器对比，而不是翻几万行日志。如果 Map 输出 100 万条、Reduce 输入只有 90 万条，问题就锁定在 Shuffle 或分区环节，范围一下就缩小了。

第三条，能不加 Combiner 就先不加。Combiner 是性能优化手段，不是必需环节。先用最朴素的实现把结果跑对，拿到基准数据，再考虑加 Combiner、调缓冲区、改分区函数。上来就一顿优化，结果出错了根本不知道是哪一步引入的。我一般的做法是保留一份"最笨但一定对"的实现作为对照组，任何优化都用小数据集跑一遍做结果比对，数字对得上才敢上生产。

第四条，中间数据路径要单独规划。Map 输出的临时数据默认落在本地磁盘，如果本地盘空间紧张，一个大作业的中间数据能把盘写满，进而引发一连串失败。规划集群时要按"最大单作业 Shuffle 数据量 × 并行任务数"估本地盘容量，别只算存储容量。这一点在早期规划阶段最容易被忽略，等到盘满了才发现，改起来就是停机扩容。

第五条，别用它做它不擅长的事。我见过用 MapReduce 做实时推荐的，每 30 秒提交一次作业，结果调度开销比计算时间还长。也见过用它做小数据量（几百 MB）的聚合，启动时间占了总耗时的九成。这类场景换个内存计算引擎或者直接用单机脚本，效果好几倍。选型的判断标准很简单：如果单机能在一分钟内算完，就不要动用分布式框架。

第六条，输出目录的清理策略要提前定。作业输出目录不允许已存在，这个设计很安全但也很烦人。生产上通常是按日期或批次建目录，由调度系统负责管理生命周期。如果这一步没设计好，跑上三个月磁盘就爆了。我的习惯是在作业配置里加一条输出保留天数的约定，配合定期的清理任务，别指望人工记得删。

第七条，失败重试次数不要盲目调大。默认 4 次重试是有道理的，如果某个任务连续失败 4 次，大概率是数据本身或者代码本身有问题，再重试 40 次也还是失败，只是把故障时间拉长了。遇到反复失败，应该立刻去看那个分片的具体数据，而不是加大重试次数碰运气。

最后分享一个我自己一直在用的习惯：每次调参之后，把关键计数器指标记到一个表格里，包括输入量、Shuffle 量、溢出次数、各阶段耗时。调过五六次之后你会发现，这个表格能帮你建立对参数和性能之间关系的直觉，比任何文档都直观。等到下次遇到新作业，看一眼数据量级，大致就能估出该给多少 Reduce、多大缓冲区，不用再从头试错。