6.824 Spring 2020 Lab1 MapReduce文档翻译

05 Sep, 2022

本文对 MIT 6.824 Lab 1:MapReduce 的说明文档进行了全文翻译。需要注意的是文中的 job 和 task，其中，job 是指整个 MapReduce 计算，表示的是任务整体，而 task 则是指一次 Map/Reduce 调用，表示的是任务局部，一个完整的 MapReduce job 由一些 Map task 和 Reduce task 组成。

翻译

引言

本实验的目标是引导您构建一个 MapReduce 系统，您需要实现两个程序：worker 和 master。其中 worker 进程负责处理文件读写操作以及调用 Map/Reduce 函数处理 Task。而 master 进程则负责为 worker 进程分配 Task 并处理崩溃的 worker 进程。您构建的系统与MapReduce 论文中所描述的系统类似。

协作政策

除了我们提供给您的代码外，您必须独立完成实验要求的代码。实现过程中杜绝参考其他同学和前几年课程实验的解决方案。允许您与同学讨论，但不允许您抄袭他们的代码。之所以设立这条规则，是因为我们认为只有独立完成实验，您的能力才会得到最大的提升。

请不要公布您的代码，也不要让它被选修 6.824 的学生通过某种方式获取到。github.com中的仓库权限默认是公开的，除非您将仓库权限设为私有，否则请勿用其存储实验代码。保存代码可以使用MIT 的 Github，但请确保您创建的是私有仓库。

软件

本实验（以及 6.824 的其他实验）使用的编程语言是Go语言，如果您不熟悉 Go 语言，其官方网站上有很多教程供您学习。我们将使用 1.13 版本的 Go 语言来评判您的代码，因此您也应该使用这一版本。另外，如果您要查看计算机中已有的 Go 语言版本，可以执行go version命令。

我们推荐您在自己的机器上完成实验，这样您就可以使用您熟悉的环境，比如工具，文本编辑器等。另外，您也可以在 Athena 上完成实验。

macOS

您可以使用Homebrew来安装 Go 语言。安装好 Homebrew 后，执行brew install go命令即可。

Linux

根据您使用的 Linux 发行版，您可以从相应的软件包库中下载最新版的 Go 语言，比如在 Ubuntu 中可以执行apt install golang命令来安装 Go 语言。此外，您还可以从 Go 语言的官方网站中手动下载二进制包。首先，确保您使用的是 64 位的 Linux 内核（uname -a会提示"x86_64 GNU/Linux"），然后执行如下命令即可：

wget -qO- https://dl.google.com/go/go1.13.6.linux-amd64.tar.gz | sudo tar xz -C /usr/local

需要确保/usr/local/bin在您的环境变量PATH中。

Windows

实验可能无法直接运行于 Windows 上。如果您敢于尝试，可以试试Windows Subsystem for Linux，并执行上述 Linux 命令。否则还是乖乖使用 Athena 吧。

Athena

您可以通过ssh {your kerberos}@athena.dialup.mit.edu命令登录公共 Athena 主机。一旦登录成功，便可通过如下命令获取 1.13 版本的 Go 语言：

setup ggo

开始

您将使用git（版本控制系统）拉取实验初始版本代码。如果您不熟悉 git，可以通过查阅Git-Book和Git User Manual自行学习。执行以下命令，即可从远端拉取 6.824 的初始实验代码：

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

在src/main/mrsequential.go中，我们为您实现了一个简单的顺序 MapReduce。由于使用的是单进程，因此该程序同一时刻仅能执行一个 Map/Reduce task。此外，我们还为您提供了多组处理不同 task 的 Map/Reduce 应用程序代码：如mrapps/wc.go中包含单词计数所需的 Map/Reduce 方法，mrapps/indexr.go中包含计算文档索引的 Map/Reduce 方法。您可以执行以下命令来运行单词计数的顺序版本 MapReduce：

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

mrsequential.go所需的输入数据来自名为pg-xxx.txt的文本文件，输出则保存在文件mr-out-0中。

您可以从mrsequential.go中随意借鉴您需要的代码，同时也可以阅读mrapps/wc.go来了解 MapReduce 的应用程序代码长啥样。

您的任务

您的任务是实现一个分布式的 MapReduce 系统，该系统由两个程序组成：master 和 worker，在系统运行期间共包含一个 master 进程和多个并行运行的 worker 进程。

在实际应用的 MapReduce 系统中，worker 进程运行在很多不同的机器上，而在本实验中，您只需要在一台机器上运行它们即可。

worker 进程和 master 进程之间使用 RPC 通信。每个 worker 进程需完成以下工作：

向 master 进程请求 task
从若干文件中读取输入数据，
执行 task，
将 task 的输出写入到若干文件中
master 进程除了为 worker 进程分配 task 外，还需要检查在一定时间内（本实验中为 10 秒）每个 worker 进程是否完成了相应的 task，如果未完成的话则将该 task 转交给其他 worker 进程。

我们为您提供了一些上下文代码。master 程序和 worker 程序的 main 函数入口分别位于main/mrmaster.go和main/mrworker.go中，不要修改这些文件，您应该将您需要实现的代码写在mr/master.go，mr/worker.go和mr/rpc.go中。

下面说明如何运行您写的代码，以单词计数应用程序为例。首先，执行以下命令以确保被构建的单词计数插件是最新版：

go build -buildmode=plugin ../mrapps/wc.go

cd 到src/main/中，运行 master 程序：

rm mr-out*
go run mrmaster.go pg-*.txt

其中pg-*.txt参数是文件路径，这些文件中保存着 master 程序的输入数据；每个文件都可以作为单个 Map task 的输入。

在其他若干窗口中，您可以运行如下命令来执行一个或多个 worker 进程：

go run mrworker.go wc.so

当 worker 进程和 master 进程执行完毕后，可以显示 mr-out-*中的内容以查看程序输出。如果您的实验代码没有错误的话，经过排序后的输出结果应与mrsequential.go的输出相同，如下所示：

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

main/test-mr.sh是我们为您提供的测试脚本。当给定pg-xxx.txt作为输入时，该脚本会检查您实现的 MapReduce 系统对两个不同类型的 Job（单词计数和计算文档索引）是否产生了正确的输出。该脚本也会检查您实现的 MapReduce 系统的 worker 进程在处理 Map task 和 Reduce task 时是否是并行运行的，以及您的系统是否能正确处理发生崩溃的 worker 进程。

如果您现在运行测试脚本，它将会挂起，因为 master 并没有被实现：

cd ~/6.824/src/main
sh test-mr.sh
*** Starting wc test.

您可以将mr/master.go中Done函数内的ret := false语句修改为 true 来使主程序立刻退出，您将会得到如下输出：

sh ./test-mr.sh
*** Starting wc test.
sort: No such file or directory
cmp: EOF on mr-wc-all
--- wc output is not the same as mr-correct-wc.txt
--- wc test: FAIL
$

测试脚本期望对于每个 reduce task，都会生成一个被命名为mr-out-X的输出文件。空的mr/master.go和mr/worker.go不会生成这些文件（也少做了很多其他事情），进而导致测试失败。

当将全部代码实现完毕后，运行测试脚本后的输出应该看起来像这样：

$ sh ./test-mr.sh
*** Starting wc test.
--- wc test: PASS
*** Starting indexer test.
--- indexer test: PASS
*** Starting map parallelism test.
--- map parallelism test: PASS
*** Starting reduce parallelism test.
--- reduce parallelism test: PASS
*** Starting crash test.
--- crash test: PASS
*** PASSED ALL TESTS
$

您也会看到一些来自 Go RPC 包输出的错误，看起来像这样：

2019/12/16 13:27:09 rpc.Register: method "Done" has 1 input parameters; needs exactly three

忽略这些信息即可。

一些规则：

map 阶段应将中间键分割为不同的桶（译者注：中间键指的是 MapReduce 论文中的 intermediate key，这句话的意思是将 map Task 的输出以某种形式保存为 reduce 函数能够读取的输入），方便后续nReduce个 reduce task 读取，而nReduce则是main/mrmaster.go传递给MakeMaster的参数；
worker 进程应把第 X 个 reduce task 的输出保存到文件mr-out-X中；
mr-out-X中每行都应该是调用一次 Reduce 函数的输出，应该按照 Go 语言的"%v %v"的格式生成，也即 key 和 value，如在main/mrsequential.go中注释"this is the correct format"的位置所示。如果您的实现和这一格式相差太多，测试脚本将会执行失败；
您需要修改的文件为：mr/worker.go，mr/master.go和mr/rpc.go。尽管您可以暂时地修改其他文件来辅助您测试，但请确保其他文件被还原为初始状态（原始版本）后您的程序仍能正常工作，我们将会使用原始版本的代码进行评判；
woker 进程应该将 Map 函数的输出（intermediate key）保存在当前目录的文件中，使得后续 worker 进程可以读取它们并将其作为 Reduce task 的输入；
当 MapReduce Job 被计算完毕后，main/mrmaster.go希望您实现的mr/master.go中的Done()方法会返回 true。这样mrmaster.go就能知道 Job 已经顺利完成，进程即可退出；
当 MapReduce job 被做完后，worker 进程就应该退出。实现这一功能的一种笨方法就是令 woker 程序检查call()函数的返回值：如果 woker 进程无法和 master 进程通信，那么 worker 进程就可以认为整个 Job 已经全被做完了，自己也就可以退出了。当然是否这么做还是要取决于您的设计，您也可以设计一个”please exit“的伪任务，当 worker 进程收到这一任务，就自动退出。

提示

如果您觉得无从下手，可以从修改mr/worker.go中的Worker()函数开始，在函数中首先实现以下逻辑：向 master 发送 RPC 来请求 task。然后修改 master：将文件名作为尚未开始的 map task 响应给 worker。然后修改 worker：读取文件并像mrsequential.go程序一样，调用 Map 方法来处理读取的数;
MapReduce 应用程序的 Map/Reduce 函数被保存在以.so结尾的文件中，在运行时使用 Go plugin 包读取；
如果您改变了mr/文件夹中的文件，并且该文件也被用到，那您需要将该文件使用类似于go build -buildmode=plugin ../mrapps/wc.go的命令重新编译成 MapReduce 插件；
本实验要求 worker 进程使用同一个文件系统，因此所有的 worker 进程必须运行于一台机器上。如果想要让 worker 程序运行在不同的机器上，那您需要为他们提供一个全局的文件系统，比如 GFS；
一个命名中间文件的合理形式是mr-X-Y，其中 X 是 Map task 的编号，Y 是 Reduce task 编号；
worker 程序处理 Map task 的代码中需要一种将中间键值对存储为文件的方式，也需要一种在处理 Reduce task 时能从文件中正确读回键值对的方式。一种可能的实现是使用 Go 语言的encoding/json包。将键值对写入 JSON 文件的代码如下：
```
  enc := json.NewEncoder(file)
  for _, kv := ... {
      err := enc.Encode(&kv)
```
从 JSON 文件中读回键值对的代码如下：
```
  dec := json.NewDecoder(file)
  for {
      var kv KeyValue
      if err := dec.Decode(&kv); err != nil {
          break
      }
      kva = append(kva, kv)
  }
```
在 worker 中处理 map Task 的部分，对于一个给定键，您可以使用ihash(key)函数（在worker.go中）来选择它属于哪一个 reduce task；
您可以将mrsequential.go中一些功能的代码拿来直接用，比如：读取 map task 的输入文件，在调用 Map 方法和调用 Reduce 方法之间对中间键值对进行排序，以及存储 Reduce 函数的输出到文件。
作为一个 RPC 服务器，master 进程将是并发的，因此不要忘记给共享资源加锁。
可以执行命令go build -race和go run -race来使用 Go 语言的 race detector，在test.sh中有一条注释为您展示如何为测试开启 race detector；
worker 进程有时需要等待，比如在最后一个 map task 处理完之前，worker 不能开始对 reduce task 的处理。实现这一功能的一种可能方案是 worker 进程周期性的向 master 请求 task，在每次请求间使用time.Sleep()阻塞一段时间。另一种可能的方案是 master 在收到 rpc 请求后额外开启一个 RPC 处理线程，在这个线程中执行循环等待（也可以使用time.Sleep()和sync.Cond），这样使得阻塞的 RPC 不会影响 master 响应其他 RPC；
master 进程无法可靠的区分崩溃的 worker 进程、活着但因某些原因停止运行的 worker 进程和正在运行但太慢导致无法使用的 worker 进程。这一问题最好的解决方案是令 master 等待一段时间，如果某个 worker 进程在这段时间（在本实验中，这段时间被设置为 10 秒）内没有完成相应的 task，就放弃继续等待并将该 task 重新分配给其他 worker。之后，master 应该假设这个 worker 进程已经死亡了（当然，它可能还活着）；
为了测试容错，您可以使用mrapps/crash.go插件，它在 Map 和 Reduce 函数中增加了随机退出功能；
为了确保没人能看到被崩溃进程写了一半的文件，MapReduce 论文提到了一个小技巧，那就是使用临时文件，一旦该文件完全写完，就自动重命名。您可以使用ioutil.TempFile创建临时文件，并使用os.Rename去自动重命名它。
test-mr.sh将会在子目录mr-tmp中运行所有的进程，因此如果有错误发生并且您想查看中间输出文件的话，可以查看这一目录中的文件。

提交步骤

注意：在正式提交前，请运行test-mr.sh。

使用命令make lab1命令打包您的实验代码并将其上传到班级的提交网站，其网址为：https://6824.scripts.mit.edu/2020/handin.py/。

第一次提交时您可能需要通过以下方式之一进行登录：1. 使用您的 MIT 证书；2. 通过 email 申请一个 API key。一旦登录成功，您的 API key（XXX）将会被显示，在控制台中输入以下命令上传 lab1 时会用到这一 API key：

cd ~/6.824
echo XXX > api.key
make lab1

注意：检查提交网站以确保您的实验代码已经成功提交！

提示：您可能提交了多次。我们将会使用时间戳来检查您的提交是否是最新的。

原文

http://nil.csail.mit.edu/6.824/2020/labs/lab-mr.html

#MapReduce