从MapReduce框架浅谈分布式计算

从MapReduce框架浅谈分布式计算

现在的数据挖掘应用，也就是常说的大数据分析，需要我们对大规模的数据进行快速高效的分析。我们先来看下面的例子：

20billions的20KB的网页有400TB，按照目前磁盘的读取速度约为30M/S，我们需要大约4个月的时间才能读完这所有的数据，而存储这些数据需要1000块普通磁盘。即便是这些问题解决了，想要分析这些数据，从中获取有用的价值，就更是难上加难了。显然我们传统的计算模式已经不能适应大数据时代的要求了。

另一方面，包括上面的例子在内的很多应用的数据都有个共同点：虽然大但是相当有规则，所以用并行的方式处理这些数据是解决问题一个很直观的方案。为了处理这类的应用，一种全新的软件栈被开发出来，这些编程系统实现并行的方式不同于以往的超级计算机，而是利用集群，即一定数量的通过以太网或者交换机连接而成的商用计算机集合。但是这样的大规模计算模式又会带来新的问题：

1、 计算如何分布到计算的节点上？

2、 怎样可以让分布式编程更简单、更高效？

3、 节点间传输数据会有很大的延时，如何解决节点间的数据传输问题？

4、 硬件故障。假设一台机器1000天（3年）发生一次故障，那么拥有1000台机器的集群每天都会发生一次故障。如何解决集群中频繁发生的硬件故障的问题？

而map-reduce软件框架正是解决了上面的所有问题，同时它具有简洁直观的特性，很快就流行起来了。

Mapreduce框架主要有两个部分，一个是map-reduce编程模型，另一个是底层新的分布式文件系统即DFS。分布式文件系统是将原始的大文件拆分成一个个小的数据块（chunks），而这些chunks远远大于传统操作系统中的磁盘块。分布式文件系统同时还保留文件的副本，因为当数据分布在大量的低廉的节点上，存储介质出现故障是常发生的，通过冗余存储来就可以将这种故障带来的损失降到最低。

在分布式文件系统之上就是各种不同的高层次的编程系统，其中最具代表性的就是mapreduce，mapreduce的实现让很多对于大数据的普通的计算都能够在集群上高效执行，而且在计算的过程中系统能自动容错。随着Mapreduce框架的迅速发展，如今甚至已经可以在mapreduce上使用更高层次的编程系统，比如SQL。

1 分布式文件系统

传统的计算都是在一个cpu上完成的，拥有自己的主存、缓存、磁盘等。过去我们常说的并行处理应用，比如科学计算，都是在那些有大量处理器和定制硬件的专用的超级计算机上完成的。后来，大规模的Web服务器的流行导致越来越多的将计算放在大量的计算节点上完成，这些计算节点都是普通的商用计算机，相比于专用的超级计算机成本降低了很多。

这种系统充分利用了并行，同时还能保证可靠性，这一点同样重要，因为它的硬件平台是由大量的独立的部分组成，任何时间任何部件都有可能发生异常。下面就来看看这种计算模式的物理组成以及基于这种物理结构的特殊文件系统。

1.1 计算节点的物理组织

这种新的并行计算，又叫作分布式计算，物理组织如下：计算节点都存放在机架上，一般一个机架上放8到64台机器，同一个机架上的节点都是通过以太网连接，而往往一个集群里有很多的机架，机架之间又是通过以太网或者交换机连接。大致的组成结构示意图如下所示：



集群中节点发生故障是很常见的事，集群中的节点数越多、网络连接越多，集群出现不能正常工作的可能性就越大。有的应用可能要花上数分钟甚至数小时在数以千计的节点上进行计算，在这么长的时间里很难保证机器不出现故障，如果每次发生故障我们都要重新开始计算，那么这个计算任务可能永远都无法成功执行。要解决这个问题，可以从如下两个方面入手：

1、 文件必须进行冗余存储。如果我们没有文件的复本，一旦节点发生意外，那么这个文件将一直无法获得直到节点被替换掉。如果我们没有备份文件，一旦磁盘报废了，磁盘上的文件将永久消失无法恢复。

2、 一个计算作业应该划分成许多的更小的任务。这样，一旦某个任务出现了问题，只需要重新执行这个小任务就行了，不影响其他的任务，更不用重新执行整个计算。

1.2 大规模文件系统的组成

要最大限度地发挥集群计算的优势，传统的文件系统显然不适合分布式环境，我们需要重新设计文件系统。新的文件系统通常称作分布式文件系统即DFS。分布式文件系统有如下三个特点：

1、文件非常大，可以到PB级别。如果是小文件，反而没有必要用分布式文件系统了

2、文件几乎不更新。文件的数据被读取用来进行计算，偶尔会有少量数据被追加到文件，但是更新并不频繁。所以像实时交易系统等都不适合采用分布式文件系统，因为虽然它们处理的数据量也很大，但是更新过于频繁。

3、读操作和文件的追加操作很多。

分布式文件系统采用全局的命名空间，文件都被划分成一个个的chunk存储，一般默认大小为64M，chunks都被备份了，默认为3份，分别放在不同的节点上，此外，保存这些复本的节点必须是其他的机架上的，这样一旦机架出了问题，复本不会全部丢失。需要注意的是，chunks的大小以及复本数都是可以用户自定义的。

分布式文件系统中的节点分为namenode和datanode，datanode才是存储数据（chunks）的节点，namenode只有一个，存储matedata，里面记录着关于文件存储的位置的全局信息，一般也有备份。用户先通过namenode获取所需要的数据的位置，然后直接连接对应的chunk读取数据。

2 mapreduce

Map-reduce是一种计算模式，它在很多系统里面都有了实现，包括谷歌自己内部的实现以及开源的Hadoop等。通过map-reduce的一种实现，你不需要复杂的操作可以进行大规模的并行计算，同时系统还能自动容错。用户只需要实现map和reduce两个函数，系统负责任务的并行执行，以及处理其中任何一个任务出错的的意外情况等。Mapreduce的执行过程大致如下：

1、 用户自定义一定数量的map任务，每个任务都会被分配一个或者多个来自分布式文件系统的chunks。这些map任务将chunks的数据转换成键值对作为输入，处理完成并输出的键值对的类型是用户在map函数中自定义的。

2、 Map任务输出的键值对会被master controller收集，然后根据键进行排序、分组，并被划分给reduce任务，默认采用哈希划分，将键相同的键值对划分给同一个reduce任务。

3、 Reduce任务每次处理一个键，通过一定的方式将同一个键的值结合到一起。具体的结合方式是由用户在reduce函数里面定义。

值得注意的是，map任务输出的键值对会被写回到本地的文件系统，而reduce任务输出的结果是保留在分布式文件系统里面，因为reduce输出的结果往往是下一个mapreduce作业的输入。

下图为mapreduce的流程图：

2.1 map 任务

我们认为map任务的输入数据（文件）都是一个个的基本元素组成的，可以是任意的类型，比如元组（tuple）或者文档。每个chunk都是由这些类似的元素组成的，而且任何元素都不可能同时存放在两个chunk里面，这样才能充分并行。一般情况下，map方法的输入也是键值对，但是键值对的键并不重要，往往都会被直接忽略掉。Map函数接收一个键值对，然后产生输出的键值对，键和值得类型都是任意的。值得注意的是，在这里，一个元素处理完后输出的键值对可以有相同的键。为了充分利用并行性，我们往往将map任务的数量设得远比机器数量高，甚至和chunks数量一样，即一个chunk对应一个map任务。因为每个chunk数据块的大小是固定的，而map任务的操作也是相同的，所以每个map任务的时间也非常接近，并行执行这些任务不会出现明显的时间不均衡，并行效率最高。map阶段执行的示意图如下：

2.2 Grouping by Key

当所有的map任务执行完，所有的键值对会根据键进行分组，拥有相同键的键值对会将值集中到一个list当中。这样就没有了重复的键，而且键值对的数量大大地减少了。然后根据reduce任务的数量，假设为r，将键值对进行哈希划分（默认划分方式），划分到r个本地临时文件当中，每一个临时文件对应唯一的一个reduce任务。每个reduce任务有多个reducer，每个reducer每次只能处理一个键。在写回本地文件系统（磁盘）保存的时候，都是先将键值对写到缓存当中，当缓存达到一定的门限的时候（比如80%），再将缓存写回到本地文件系统中，这样可以大大提高写的速度。

2.3 combiners

然后我们会发现上面的处理有一个问题，同一个map任务输出的键值对的键有很多的重复的，使得中间临时键值对数量太大，如果我们能够在键值对输出之前就把相同键值对进行合并，那么输出的键值对数量会大大的减少，同时后续的reduce的计算任务也会降低很多。其实我们可以在一个map任务的内部添加一个结合器（combiner），将一个map任务产生的键相同的键值对进行合并。这样提高了效率，同时对reduce最终输出的结果并没有影响。虽然这样保证了每个map任务输出的键不会重复，但是combiner并不能合并不同任务的键值对，所以最后所有map任务结束后，键依然会有大量重复，只是键值对数量少了很多。

2.4 reduce任务

Reduce函数的输入是一个键值对，其中值是相关值组成的一个list，输出可以是一个或多个键值对，键值对的类型也可以不同于输入的键值对，但是一般情况下是相同的。输出的键值对不写回本地文件系统，而是直接保存在分布式文件系统。相比于map任务，我们应该设置多少个reduce任务合适呢？如果我们把r也设置得很大，一方面就意味着划分阶段，会产生更多的临时文件，这些文件都是存储在本地的文件系统，读写效率会限制系统的性能；另一方面，如果r设置的很大，那么每个reduce任务所处理的键就会减少，我们知道map阶段结束后，每个键所对应的值得数量分布非常不均匀，有的键值对很多，有的键值对很少，reduce处理的键越少，这种不均匀就体现的越明显，相反，如果r很小，每个reduce处理的键就会变多，键的不均匀分布就被削弱了，每个reduce任务执行的时间的差异也就不会那么大了。所以reduce任务的数量不宜过大，很多时候我们甚至都只设置一个reduce任务。



到这里，mapreduce框架及其原理已经分析得差不多了，后续还会有关于hadoop以及sparkg更深入的研究，请关注我的博客。