6.5840 实验 1：MapReduce

介绍

在这个实验中，你将构建一个 MapReduce 系统。 你将实现一个调用应用程序 Map 和 Reduce 函数并处理文件读写的 worker 进程， 以及一个将任务分配给 worker 并处理 worker 故障的 coordinator 进程。 你将构建类似于 MapReduce 论文 中的系统。 （注意：本实验使用 "coordinator" 代替论文中的 "master"。）

入门

你需要 配置 Go 来完成实验。

使用 git（一个版本控制系统）获取初始实验软件。 要了解更多关于 git 的知识，请查看 Pro Git 书 或 git 用户手册。

$ git clone git://g.csail.mit.edu/6.5840-golabs-2026 6.5840
$ cd 6.5840
$ ls
Makefile src
$

我们在 src/main/mrsequential.go 中提供了一个简单的串行 mapreduce 实现。它在单个进程中依次运行 map 和 reduce。我们还提供了几个 MapReduce 应用程序：mrapps/wc.go 中的词频统计，以及 mrapps/indexer.go 中的文本索引器。你可以按如下方式串行运行词频统计：

$ cd ~/6.5840
$ cd src/main
$ go build -buildmode=plugin ../mrapps/wc.go
$ rm mr-out*
$ go run mrsequential.go wc.so pg*.txt
$ sort mr-out-0
A 509
ABOUT 2
ACT 8
ACTRESS 1 
...

mrsequential.go 将其输出留在文件 mr-out-0 中。输入来自名为 pg-xxx.txt 的文本文件。

你可以随意从 mrsequential.go 中借用代码。你还应该查看 mrapps/wc.go 以了解 MapReduce 应用程序代码的样子。

在本实验及所有其他实验中，我们可能会对我们提供的代码进行更新。为了确保你能获取这些更新并使用 git pull 轻松合并，最好将我们提供的代码保留在原始文件中。你可以按照实验说明在我们提供的代码基础上添加内容；只是不要移动它。你可以将自己的新函数放在新文件中。

你的任务

我们提供了一些初始代码。coordinator 和 worker 的 "main" 例程在 main/mrcoordinator.go 和 main/mrworker.go 中；不要修改这些文件。你应该将实现放在 mr/coordinator.go、mr/worker.go 和 mr/rpc.go 中。

以下是在词频统计 MapReduce 应用程序上运行你的代码的方法。首先，构建词频统计插件：

$ cd main
$ go build -buildmode=plugin ../mrapps/wc.go

$ rm mr-out*
$ go run mrcoordinator.go sock123 pg-*.txt

在一个或多个其他窗口中，运行一些 worker：

$ go run mrworker.go wc.so sock123

$ cat mr-out-* | sort | more
A 509
ABOUT 2
ACT 8
ACTRESS 1 
...

我们提供了所有用于评分的测试。测试的源代码在 mr/mr_test.go 中。你可以在 src 目录中运行测试：

$ cd src
$ make mr
...

如果你现在运行测试，它们会在第一个测试中挂起：

$ cd ~/6.5840/src
$ make mr
...
cd mr; go test -v -race 
=== RUN   TestWc
...

你可以在 mr/coordinator.go 的 Done 函数中将 ret := false 改为 true，使 coordinator 立即退出。然后：

$ make mr
...
=== RUN   TestWc
2026/01/22 14:56:24 reduce created no mr-out-X output files!
exit status 1
FAIL    6.5840/mr       4.516s
make: *** [Makefile:44: mr] Error 1
$

测试期望在名为 mr-out-X 的文件中看到输出，每个 reduce 任务一个。mr/coordinator.go 和 mr/worker.go 的空实现不会生成这些文件（也基本不做其他任何事情），所以测试失败。

当你完成后，测试输出应该如下所示：

$ make mr
...
=== RUN   TestWc
--- PASS: TestWc (8.64s)
=== RUN   TestIndexer
--- PASS: TestIndexer (5.90s)
=== RUN   TestMapParallel
--- PASS: TestMapParallel (7.05s)
=== RUN   TestReduceParallel
--- PASS: TestReduceParallel (8.05s)
=== RUN   TestJobCount
--- PASS: TestJobCount (10.04s)
=== RUN   TestEarlyExit
--- PASS: TestEarlyExit (6.05s)
=== RUN   TestCrashWorker
2026/01/22 14:58:14 *re*-starting map ../../main/pg-tom_sawyer.txt 0
2026/01/22 14:58:14 *re*-starting map ../../main/pg-metamorphosis.txt 2
2026/01/22 14:58:39 *re*-starting map ../../main/pg-metamorphosis.txt 2
2026/01/22 14:58:40 map 2 already done
2026/01/22 14:58:45 *re*-starting reduce 0
--- PASS: TestCrashWorker (40.18s)
PASS
ok      6.5840/mr       86.932s
$

根据你终止 worker 进程的策略，你可能会看到如下错误：

2026/02/11 16:21:32 dialing:dial unix /var/tmp/5840-mr-501: connect: connection refused

一些规则：

• map 阶段应该将中间键分配到 nReduce 个 reduce 任务的桶中，其中 nReduce 是 reduce 任务的数量——即 main/mrcoordinator.go 传递给 MakeCoordinator() 的参数。每个 mapper 应该创建 nReduce 个中间文件供 reduce 任务使用。
• worker 实现应该将第 X 个 reduce 任务的输出放在文件 mr-out-X 中。
• mr-out-X 文件应该每行包含一个 Reduce 函数的输出。该行应该使用 Go 的 "%v %v" 格式生成，传入键和值。请查看 main/mrsequential.go 中标注为 "this is the correct format" 的行。如果你的实现与此格式偏差太大，测试将失败。
• 你可以修改 mr/worker.go、mr/coordinator.go 和 mr/rpc.go。你可以临时修改其他文件进行测试，但请确保你的代码在原始版本下能正常工作；我们将使用原始版本进行测试。
• worker 应该将中间 Map 输出放在当前目录的文件中，以便 worker 稍后可以将其作为 Reduce 任务的输入读取。
• main/mrcoordinator.go 期望 mr/coordinator.go 实现一个 Done() 方法，当 MapReduce 作业完全完成时返回 true；此时 mrcoordinator.go 将退出。
• 当作业完全完成时，worker 进程应该退出。一种简单的实现方式是使用 call() 的返回值：如果 worker 无法联系 coordinator，它可以假设 coordinator 已退出因为作业已完成，因此 worker 也可以终止。根据你的设计，你可能还会发现让 coordinator 给 worker 分配一个 "请退出" 的伪任务会很有帮助。

提示

• 指导页面有一些开发和调试的技巧。
• 一种入门方式是修改 mr/worker.go 的 Worker()，向 coordinator 发送 RPC 请求任务。然后修改 coordinator 以返回一个尚未开始的 map 任务的文件名。然后修改 worker 以读取该文件并调用应用程序 Map 函数，如 mrsequential.go 中所示。
• 应用程序的 Map 和 Reduce 函数在运行时使用 Go 的 plugin 包从文件名以 .so 结尾的文件中加载。
• 如果你修改了 mr/ 目录中的任何内容，你可能需要重新构建你使用的任何 MapReduce 插件，使用类似 go build -buildmode=plugin ../mrapps/wc.go 的命令。make mr 会为你构建插件。你可以使用 make RUN="-run Wc" mr` 运行单个测试，它将 "-run Wc" 传递给 go test，并从 mr/mr_test.go 中选择任何匹配 Wc 的测试。
• 本实验依赖于 worker 共享文件系统。当所有 worker 运行在同一台机器上时这很简单，但如果 worker 运行在不同机器上则需要像 GFS 这样的全局文件系统。
• 中间文件的合理命名约定是 mr-X-Y，其中 X 是 Map 任务编号，Y 是 reduce 任务编号。
• worker 的 map 任务代码需要一种将中间键/值对存储在文件中的方式，以便在 reduce 任务期间能正确读回。一种方式是使用 Go 的 encoding/json 包。将键/值对以 JSON 格式写入打开的文件：

    enc := json.NewEncoder(file)
    for _, kv := ... {
      err := enc.Encode(&kv)
  

  读回这样的文件：

    dec := json.NewDecoder(file)
    for {
      var kv KeyValue
      if err := dec.Decode(&kv); err != nil {
        break
      }
      kva = append(kva, kv)
    }
  

• worker 的 map 部分可以使用 ihash(key) 函数（在 worker.go 中）来选择给定键对应的 reduce 任务。
• 你可以从 mrsequential.go 中借用一些代码来读取 Map 输入文件、在 Map 和 Reduce 之间排序中间键/值对，以及将 Reduce 输出存储在文件中。
• coordinator 作为 RPC 服务器将是并发的；不要忘记锁定共享数据。
• worker 有时需要等待，例如 reduce 在最后一个 map 完成之前无法开始。一种方式是让 worker 定期向 coordinator 请求工作，在每次请求之间使用 time.Sleep() 休眠。另一种方式是让 coordinator 中相关的 RPC 处理程序有一个等待循环，使用 time.Sleep() 或 sync.Cond。Go 为每个 RPC 在自己的线程中运行处理程序，所以一个处理程序在等待不会阻止 coordinator 处理其他 RPC。
• coordinator 无法可靠地区分崩溃的 worker、存活但因某种原因停滞的 worker、以及正在执行但太慢而无法有用的 worker。你能做的最好的方式是让 coordinator 等待一段时间，然后放弃并将任务重新分配给另一个 worker。在本实验中，让 coordinator 等待十秒；之后 coordinator 应该假设 worker 已死亡（当然，它可能并没有）。
• 如果你选择实现备份任务（Backup Tasks，第 3.6 节），请注意我们测试了你的代码在 worker 不崩溃地执行任务时不会调度多余的任务。备份任务应该只在相对较长的一段时间后（例如 10 秒）才被调度。
• 要测试崩溃恢复，你可以使用 mrapps/crash.go 应用程序插件。它在 Map 和 Reduce 函数中随机退出。
• 为了确保在崩溃情况下没有人观察到部分写入的文件，MapReduce 论文提到了使用临时文件并在完全写入后原子性地重命名它的技巧。你可以使用 ioutil.TempFile（如果你运行的是 Go 1.17 或更高版本则使用 os.CreateTemp）来创建临时文件，并使用 os.Rename 来原子性地重命名它。
• Go RPC 只发送名称以大写字母开头的结构体字段。子结构体也必须有大写字母开头的字段名。
• 调用 RPC call() 函数时，reply 结构体应包含所有默认值。RPC 调用应该如下所示：

    reply := SomeType{}
    call(..., &reply)
  

  不要在调用前设置 reply 的任何字段。如果你传递了具有非默认字段值的 reply 结构体，RPC 系统可能会默默返回不正确的值。

不计分挑战练习

实现你自己的 MapReduce 应用程序（参见 mrapps/* 中的示例），例如分布式 Grep（MapReduce 论文的第 2.3 节）。

让你的 MapReduce coordinator 和 worker 在单独的机器上运行，就像实际部署中那样。你需要设置 RPC 通过 TCP/IP 而不是 Unix socket 进行通信（参见 Coordinator.server() 中被注释掉的行），并使用共享文件系统读写文件。例如，你可以在 MIT 的多个 Athena 集群机器上使用 ssh，它们使用 AFS 共享文件；或者你可以租用几个 AWS 实例并使用 S3 进行存储。