编写mapreduce程序实现对输入文件的词频统计排序_MIT 6.824 Lab 1: MapReduce (1)

0 写在前面的话

最近花了一段时间完成了2018 Spring MIT 6.824分布式系统公开课的实验，写一些文章记录下自己的一点心得体会。对于6.824的第一个实验MapReduce，我将分成(1)和(2)两篇文章进行讲解。本篇文章主要讲的是MapReduce的理论知识，也就是MapReduce论文中的内容，具体的实验内容和代码，将在文章(2)中讲解。为了便于理解，这里省去了论文中的一些细节，只讲了MapReduce的主干内容。

1 前言

对于大多数互联网公司来说，数据是一项非常非常宝贵的资产，但是在面对海量数据时，单机往往难以处理，需要使用若干台机器去协同处理海量数据。然而，单机上简单的问题在多台机器协同处理时往往并不简单。Dean等[1]提出了一种编程模型–MapReduce，用来解决多机协同处理海量数据的难题。

2 背景

为了方便读者理解MapReduce，我们先引入一个问题：某互联网公司用数量巨大的文本数据，大约有PB级，如何统计这些文本数据中每个单词出现的次数？假设用一台计算机去处理这些数据的话，很显然不可能将所有数据都读入到内存中，只能每次读取一定量的数据，然后统计读入的数据中每个单词出现的次数，直到将所有数据读取完毕。这种批量处理方法效率非常低下，因为每次只能处理一小部分数据，是一种串行的处理方法。我们考虑使用并行的方法，使用多台计算机协同处理这些数据，该如何做呢？原理很简单，首先将PB级的数据分成若干部分，每部分小到可以读取到计算机内存中，将这些数据“分散”到每台计算机上去统计单词出现的次数，然后将所有计算机的统计结果“汇总”起来，就得到了最终结果。这里的“分散”就对应了MapReduce中的“map”，“汇总”对应了“reduce”。

3 理论知识

这里我们只介绍MapReduce的主要知识，一些细节的东西就不做这里说明。我们分成两部分介绍，一部分是基于MapReduce提供给用户的接口，这一部分是需要用户编写代码的；另一部分介绍MapReduce是如何并行执行用户的指令的，这一部分是由MapReduce库函数完成的。

3.1 用户接口

数据处理任务繁多，MapReduce提取了这些繁杂任务的共同点，抽象出了如下编程模型：

对于一项数据处理任务，可以用一系列键/值对表示输入，一系列键/值对对表示输出。 MapReduce用Map函数和Reduce函数表示数据处理过程。

Map函数，由用户根据需求实现，对实现过程没有任何要求，但是要求：(1)输入是一个键/值对；(2)输出是一系列中间 (intermediate) 键/值对。接下来的处理，不需要用户编码实现。对于输出的一系列中间键/值对，MapReduce库函数把具有相同键I的中间键/值对划分为同一组，然后将这些具有相同键I的中/间键值对传给Reduce函数。

Reduce函数，同样由用户根据需求实现。如前所述，Reduce函数的输入是一系列具有相同键的中间键/值对，也就是输入的是一个中间键I ，以及I 对应的一系列值。Reduce函数合并这些值，生成新的值。通常调用一次Reduce函数，输出1个或0个值。

上面对Map函数和Reduce函数的描述可能过于抽象，我们拿前面提到的词频统计为例，实现能够统计词频的Map函数和Reduce函数，如代码1和代码2所示。

map(String key, String value):
  //key: document name
  //value: document contents
  for each word w in value:
    EmitIntermidate(w,"1");

代码1： Map函数代码

reduce(String key, Iterator values):
  //key: a word
  //values: a list of counts
  int result=0;
  for each v in values:
    result+=ParseInt(v);
  Emit(AsString(result));

代码2：Reduce函数代码

代码1输入一个文档名和该文档对应的内容，然后遍历文档的所有单词，对遍历到的每个单词都生成一个中间键/值对 (w,"1")。需要注意的是，一篇文档中可能存在多个重复的单词，对每一个重复的单词，都会新生成一个的中间键值对。如某篇文档中的单词为：

{"code", "map", "reduce", "code"}

则map函数输出的一系列中间键值对为:

[<"code", "1">, <"map", "1">, <"reduce", "1">, <"code", "1">]

其中文档中"code"出现了两次，也生成了两个中间键/值对<"code", "1">。细心的读者可以发现，输入参数key在这里并没用到，实际上大多时候map函数中的参数key都不会用到。

代码2中，参数Iterator values可以简单认为是键key对应的值的列表，如输入的key是"code"，则values为["1","1"] ，最终reduce函数输出的结果为"2"，这正是文档中单词“code”出现的次数。

3.2 并行执行过程

图1 MapReduce并行执行过程

MapReduce库函书如何将用户编写的Map函数和Reduce函数并行执行呢，我们需要分开来说。对于Map函数，我们将输入的数据分成M 份，对于每一份数据，都可以由一台机器执行Map函数来处理，也即可以由M 台机器并行执行Map函数。对于Reduce函数，我们也仿照Map函数的并行方法，将Reduce函数的输入数据分成R份，然后再由R台机器并行执行Reduce函数。由于Reduce函数输入数据是Map函数输出的中间键/值对，因此也就是将Map函数生成的所有中间键/值对分成R份。需要注意的是，Map函数输入数据分成M 份，理论上是可以随便分的，但是Reduce函数的输入数据分成R份，却不是随便分的，应该根据中间键/值对的键来分成R份，比如 hash(key) mod R的值相同的中间键值对分为同一份，具体原因，我们将在实验解析部分给出。整体流程如图1所示。具体过程如下（(图1中的数字和下面的一致)：

1. MapReduce库函数将输入数据分成M 份，每份数据大小通常在16 - 64MB之间。然后在众多计算机中选出一台计算机作为Master。
2. 除了Master，剩下的计算机都称为Worker。Master负责给这些Workers分配任务，共有M 个Map任务和R个Reduce任务需要分配。Master选择空闲的Worker分配任务 1 。
3. 分配到Map任务的Worker读取相应的数据份，然后将这份数据中的键/值对传递给用户编写的Map函数，再将生成的中间键/值对缓存到内存中。
4. 对于内存中的中间键/值对，定期使用partition函数(如 hash(key) mod R )分成R份，存储到本地计算机磁盘上。每一份中间键/值对在磁盘上的地址报告给Master，因为在分配Reduce任务时需要用到这个地址。
5. 当一个分配到Reduce任务的Worker接收到Master传来的这些中间键/值对的地址后，这个Worker使用远程过程调用读取这些中间键/值对。当这个Worker读取完所有的中间键/值对后，根据键对这些中间键/值对排序，从而将所有键相同的中间键/值对聚集到了一起。
6. 分配到Reduce任务的Worker遍历这些排序了的中间键/值对，然后某个键对应的一系列值传递给用户编写的Reduce函数，每一个Reduce任务，都会生成一个文件保存Reduce函数的输出结果。
7. 当所有Map任务和Reduce任务执行完成后，将生成R个结果文件，MapReduce库函数返回到用户程序中。

4 结语

以上就是MapReducce主要的理论知识，我将在下一篇文章中讲解6.824 MapReduce的实验内容。