MapReduce 设计流程

背景：2004 年，Google 需要在 TB 级别的数据上进行大量计算（比如对所有网页创建索引，非常费时），Google 希望将对大量数据的大量运算并行跑在几千台计算机中，快速完成计算。

按照传统方式，工程师在编写业务代码时，还需要花大量时间编写分布式的相关代码。因此 Google 需要一种框架，使得普通工程师只需编写核心代码，而无需考虑分布式计算的工作。因此 MapReduce 出现了。

以下是一个完整的 MapReduce Job。

• Job。整个 MapReduce 计算称为 Job。
• Task。每一次 MapReduce 调用称为 Task。
  所以，对于一个完整的 MapReduce Job，它由一些 Map Task 和一些 Reduce Task 组成。

流程（以单词计数器为例）

需要说明的是，课程 lab1 所实现的 MapReduce 框架与论文中的细节存在一定出入，因此本文的流程并非与课程实验一一对应。我在本文和 MIT6.5840 Lab 1：MapReduce 中分别阐述了两次流程：

• 本文忠于 MapReduce 论文的原始设计，侧重于流程设计而非实现细节
• 另一篇文章则在保持流程整体一致的前提下，为满足课程实验的要求做了相应适配。侧重于实现的细节与决策而非流程设计。

前提：MapReduce 假设输入数据是静态的、全量的。必须 Map workers 将所有输入数据全都处理结束后，才能进行下一步（reduce）。

换句话说，它采用批处理（batch）方式，而不是流（streaming）

初始数据：两份文本文件：File1: "hello world"，File2: "hello mapreduce"

设置初始参数：

• M=2（2 个 Map 任务）
• R=2（2 个 Reduce 任务），
• 哈希分区函数：hash(key) mod R）（来决定 Key 发往哪个 Reduce）
• 假设哈希值：hash("hello")%2=0, hash("world")%2=1, hash("mapreduce")%2=0,

阶段一：输入分片（Input Splits）与数据本地化调度 (即任务分配)

1. 物理存储分块（GFS）：输入大文件在写入存储系统时，已被分布式文件系统自动按固定大小（如 64MB）物理切分为多个数据块（Block），并分散存储在不同节点的本地磁盘上
   • Block 0 在 Worker A 的本地硬盘。内容为 "hello world"，
   • Block 1 在 Worker B 的本地硬盘，内容为 "hello mapreduce"）。

2. 逻辑分片映射（Master）：Master 向文件系统查询，在内存中将这些 Block 映射成逻辑上的 Input Splits（即 Map Task）。以 Hadoop 为例，InputSplit 对象的真实数据结构如下：

public class FileSplit extends InputSplit {
    private Path file;         // 文件路径，例如: hdfs://cluster/logs/2026-08-14.log
    private long start;        // 本分片的起始字节偏移量 (Offset)，例如: 67108864 (64MB)
    private long length;       // 本分片的字节长度 (Length)，例如: 67108864 (64MB)
    private String[] hosts;    // 存储该 Block 副本的节点 IP (用于数据本地化调度)
}

3. Master 遵循数据本地化原则，将 Map Task 1 分配给 Worker A，将 Map Task 2 分配给 Worker B。数据无需跨网络传输，直接从本地磁盘读取。

阶段二：Map 阶段

Map Task 1（Worker A）的处理流程（要注意执行主体。MR 框架是底层框架的行为，Map 函数是用户的行为）

1. MR 框架将 Block 数据块解析为键值对：key="0"（表示偏移量）,value="hello world"
2. MR 框架执行用户定义的 Map 函数，由 Map 遍历每个单词，并 emit 中间键值对：
   • EmitIntermediate("hello", "1")
   • EmitIntermediate("world", "1")
   • 这些中间键值对先以未分区、未排序的形式写入内存缓冲区（In-Memory Buffer）。此时，框架仅负责顺序存储，不计算分区，也不触发排序。
3. 当内存缓冲区达到阈值后，MR 框架的后台线程会启动溢写操作，将缓冲区内所有中间键值对进行分区，接着在每个分区内部对所有中间键值对进行排序，最后写入本地磁盘。同一分区的多个溢写文件会在 Map 任务结束前会进行归并排序，写入中间文件
   • hash("hello") % 2 = 0 → 进入分区 0
   • hash("world") % 2 = 1 → 进入分区 1

MapReduce 的代码实现是闭源的，上述流程参照了 Hadoop，即主流的、最符合论文思想的开源实现。

一个 Map 任务产生一个中间文件和一个索引文件；一个中间文件包含多个分区，分区位置由索引文件维护；每个分区内部是局部有序的

Map Task 2（Worker B）的处理流程：略

Map 任务完成：

• Map Worker 将本地中间文件的路径，以及索引文件中各分区的起始偏移量和长度报告给 Master。
• Master 存储这些信息。当调度 Reduce 任务时，Master 将属于该分区的所有 <文件路径, 偏移量, 长度> 元组通知给 Reduce Worker。
• Reduce Worker 收到通知后，主动发起 RPC 请求，去 Map Worker 的本地磁盘拉取（Pull）指定偏移量和长度的数据块。

阶段三：Shuffle（混洗）与 Sort（排序）

Reduce Worker-0（负责处理分区 0，即 R=0）：

1. 通过网络 RPC，从 Worker-A 和 Worker-B 的本地磁盘拉取所有属于分区 0 的中间数据：
   • 来自 Worker-A：<"hello", "1">
   • 来自 Worker-B：<"hello", "1"> 和 <"mapreduce", "1">
2. 排序：Reduce 0 将拉取到的两股数据流合并成一个按 Key 全局排序的大文件。合并后的内存/磁盘数据呈现为：
   [("hello", 1), ("hello", 1), ("mapreduce", 1)] （按字典序排列，相同的 Key 紧挨着）。
3. 分组（预处理）：Reduce Worker 在排序完成后，遍历排序好的列表，将相同 Key 的 Values 封装成一个迭代器（Iterator），即转化为逻辑上的 (Key, list(Values)) 格式，例如 ("hello", [1,1]) 和 ("mapreduce", [1])

阶段四：Reduce 阶段

1. Reduce Worker-0 执行用户定义的 Reduce 函数：
   • 对于键 "hello"，传入迭代器 [1,1]，在 Reduce 函数中累加：result = 1 + 1 = 2。发出最终结果 Emit("hello", 2)，将键值对写入临时文件
   • 对于键 "mapreduce"，传入迭代器 [1]，累加：result = 1。发出最终结果 Emit("mapreduce", 1)，将键值对追加（Append）写入同一个临时文件。
2. 当所有 Key 都遍历完毕，执行一次原子重命名（Atomic Rename）操作，正式生成为最终输出文件 part-r-00000（Hadoop 的惯用命名）。

阶段五：结果呈现

Master 唤醒用户程序。用户得到 R=2 个输出文件，内容分别为：

part-r-00000（由 Reduce 0 产出）：

hello 2
mapreduce 1

part-r-00001（由 Reduce 1 产出）：

world 1

关注点

MR 如何优化网络通信？

1. 数据本地性优化（Locality Optimization）
   • Master 会优先将 Map 任务调度到包含该输入数据副本的机器上，减少网络传输。
2. Map 端本地聚合（Combiner）
   • Map 任务在将数据写入本地磁盘之前，先对相同 Key 的中间结果进行合并。减少了跨网络的中间数据量。

MR 如何实现负载均衡（load balance）？

1. 细粒度任务划分（Fine-grained Task Granularity）
   • M 和 R 远大于集群中的 worker 数量。这样每个 worker 执行多个任务，性能快的机器执行更多任务、慢的执行更少。
2. 动态调度（Dynamic Scheduling）
   • Master 维护所有任务的状态（idle/in-progress/completed）。它会动态地将空闲任务分配给空闲的 worker。
3. 备份任务（Backup Tasks）
   • 当计算接近完成时，Master 会为剩余未完成的 in-progress 任务启动备份执行（即 speculative execution）。
   • 无论是主执行还是备份执行，谁先完成，该任务就标记为完成。这有效解决了因磁盘坏道、CPU 争用等导致的个别机器“拖后腿”问题，直接均衡了最终完成时间。

MR 如何实现容错（fault tolerance）？（有哪些可能出现的错误？如何解决？）

1. Worker 故障（Worker Failure）

   • 表现：机器宕机、网络不可达、进程崩溃。
   • 检测方式：Master 会周期性 ping 每个 worker。若在一段时间内未收到响应，则将该 worker 标记为 failed。
   • 处理方式：
     • 已完成（completed）的 Map 任务：需要重新执行。因为 Map 的输出存储在故障机器的本地磁盘上，无法访问。这些任务会被重置为 idle 状态，重新调度给其他 worker。
     • 已完成（completed）的 Reduce 任务：不需要重新执行。因为 Reduce 的输出存储在全局文件系统（GFS）中，其她机器可访问。
     • 正在执行（in-progress）的 Map 或 Reduce 任务：同样重置为 idle，重新调度。

2. Master 故障（Master Failure）

   • 处理方式：论文提到，Master 会定期将数据结构写为检查点（checkpoint）。如果 Master 进程死亡，可以从最后一个检查点恢复。
   • 当前实现：但由于只有一个 Master，其故障概率很低，论文中实际的实现是如果 Master 失败，则中止整个 MapReduce 计算，由客户端决定是否重试。

3. 语义一致性保证（处理重复执行和原子性）

   • 由于上述故障会导致任务重试（尤其是 Map 任务可能被多个 Reduce 读，或同一 Reduce 被执行多次），MapReduce 依赖原子提交（atomic commit）来保证结果正确性。
   • 机制（论文 3.3 节）：
     • 每个 Map 任务产生 R 个临时文件（对应 R 个 Reduce 分区），完成后将文件名通知 Master。
     • 每个 Reduce 任务将其输出写入临时文件，任务完成后原子性地 rename 为最终输出文件。
     • 依赖底层文件系统（GFS）的原子 rename 操作，确保即使同一 Reduce 任务在多台机器上执行多次，最终文件系统状态中也只有一次成功执行的结果。

4. 用户代码导致的“坏记录”（Bad Records）

   • 表现：用户 Map 或 Reduce 函数在处理某条特定记录时，因 bug 导致崩溃（如段错误、总线错误），且反复执行都会崩。
   • 解决方式（论文 4.6 节）：
     • Worker 安装信号处理器（捕获 segmentation violations 等）。
     • 调用用户函数前，记录当前处理记录的序列号。
     • 若崩溃，信号处理器会发送包含该序列号的 UDP “last gasp” 包给 Master。
     • 当 Master 发现同一条记录失败超过一次，会在后续重新执行该任务时指示 worker 跳过该记录，从而让整个作业继续推进（适用于统计分析等可容忍少量丢失的场景）。

5. “落后者”（Stragglers）

   • 表现：磁盘坏道（读速从 30MB/s 降到 1MB/s）、CPU 缓存被禁用、被其他任务抢占资源等。
   • 解决方式：即上述的 “备份任务（Backup Tasks）”机制（3.6 节），本质上是一种针对“性能故障”的容错手段。