MapReduce 是 Google 的 Jeff Dean 等人在 2004 年提出的大数据处理框架。它将数据拆分为多块，在多个机器上并行处理。原本需要高性能计算资源的数据处理任务，通过 MapReduce 可以放到多个廉价的机器上并行处理。

在 MIT 6.824 分布式系统课程中，第一个实验便是用 Go 实现 MapReduce。

下图描绘了 MapReduce 的过程：

以计算单词个数为例，它的 Map 和 Reduce 函数可以这么表示：

更多的细节在论文 MapReduce: Simplified Data Processing on Large Cluster 中，对 MapReduce 有着很详细的解释。

下面将基于 MIT 6.824 的实验部分，讲述一下我是如何实现基本的 MapReduce 的。现在网上大多数的实验版本还停留在几年前，但 2020 Spring 的实验代码结构有了很大的变化，相对更加自由。

首先直接克隆 MIT 6.824 的项目：

项目概览：所有的文件都在 src 文件夹下面。按照实验页面描述的可以执行已经给好的 sequential 的 MapReduce。示例是执行 wc.go 的统计单词数的任务。

这个 Lab 是实现一个分布式的 MapReduce，由 master 和 worker 组成，它们将通过 RPC 进行通讯。

在这个 Lab 中，我们只需要修改 src/mr 目录下的 master.go、rpc.go 和 worker.go。

这个 Lab 的切入口是 worker.go，我们在 Worker 函数中，需要不断请求任务，master 根据当前状态分配任务。因此 worker 并不需要管理状态，只需要根据 RPC 返回的结果执行相应的任务。

对于 master，我们需要所有输入文件名、Reduce 任务数量 R。因为 Map 任务的数量等于输入文件的数量 M，所以可以维护一个长为 M 的数列来管理所有 Map 任务的状态。同理也可以用长为 R 的数列管理 Reduce 任务。因而主要的 master 数据结构可以如下构造：

对于每个任务，自定义了一个 MapReduceTask 的结构体封装对应的数据。根据上面的需求分析，每个任务需要有：

RPC 消息的定义非常简单，worker 向 master 请求时有两种情况：

而 master 回复时则只需要指定 Task 以及告诉总的 Reduce 任务数量。

因此 master 的逻辑便是：

Master 结构初始化时，得到所有待处理文件 files 以及 Reduce 结果数量 nReduce。

将待处理文件顺序加入到 mapTasks 中，并初始化每个任务状态为 Unassigned。同样的，将 nReduce 个 Task 加入 reduceTasks 中

定义 RPC 处理函数 WorkerCallHandler(args *MapReduceArgs, reply *MapReduceReply) error，worker 的 RPC 请求将由这个函数处理

判断请求类型 RequestTask/FinishTask

如果是请求任务，先通过 mapFinished 判断是否 Map 阶段已完成，若未完成，则在 mapTasks 中选择一个 Unassigned 或者超时的任务；对于 Reduce 阶段，也是一样的。如果所有 Map 任务都已经被分配，并且没有完全结束，则返回一个 Wait 任务让 worker 等待

如果是完成任务，则首先判断 worker 返回的任务 Timestamp 是否一致（因为会有任务超时被重新分配，所以时间戳会不一致）。接着将对应的任务标记为 Finished，并判断是否所有的任务都已经 Finished。

需要注意的：

对于 worker，只需要根据分配的任务执行相应的函数。具体的 Map 和 Reduce 任务可以参考 mrsequential.go：

在实现好之后通过 src/main/test-mr.sh 执行所有的测试：

参考：