玩扑克牌学大数据：小白也能读懂的MapReduce工作原理

大数据处理目前比较流行的是两种方法，一种是离线处理，一种是在线处理。

在互联网应用中，不管是哪一种处理方式，其基本的数据来源都是日志数据，例如对于web应用来说，可能是用户的访问、点击日志等。

如果对于数据的分析结果在时间上有比较严格的要求，则可以采用在线处理的方式。比如双十一的成交额需要实时动态进行更新的，对于这种情况，可以使用Spark、Storm等进行在线处理。

当然，如果只是希望得到数据的分析结果，对处理的时间要求不严格，就可以采用离线处理的方式，比如我们可以先将日志数据采集到HDFS中，之后再进一步使用MapReduce来对数据进行分析。

MapReduce是Hadoop平台的基础组件之一，它是一个分布式的计算框架，用于大数据的离线计算，和HDFS、YARN搭配使用。

MapReduce主要分为切片File、Map、shuffle和Reduce四个阶段。本文主要以一个扑克牌的类比案例，来详细聊聊MapReduce的工作原理。

MapReduce的原理和洗牌一样简单

IT的智慧体现在算法上，而算法的灵感就来源于生活。

想要理解MapReduce做离线计算到底是怎么回事，也得从生活中找到答案。

现在有两幅扑克牌无序的堆成一摞放在桌子上，四个人玩牌，玩牌之前他们要看看每一种花色的牌是否完整？请问大家，用什么方式最快？

——正确答案：把牌分成4摞，每个人拿一摞，拿到牌之后每个人把自己手里的牌按照红黑花梅的花色分堆，大家都分好之后再按照花色把四个人手中的牌整合到一起。红桃一摞，黑桃一摞，以此类推。

到此为止，我们就已经成功的将两幅混乱的扑克牌按花色分成四摞了，然后每个人去这四摞牌里面拿到一种花色，从最小排到最大，去清点自己手里的花色是否完整。

其实这就是一个离线计算的过程。

离线计算的过程分为两个阶段，首先是Map的阶段，就是我们最开始的时候把混乱的扑克牌分成4摞，每个人去拿一摞将其一张张分开的过程。分开之后，我们检查花色，并且按照花色将其分堆，这就是一个Partition的过程。

Partition完成之后呢？我们每个人的手上保存了4摞分好了花色的牌，这时候就进入Reduce阶段，进入Reduce阶段之后又做些什么呢？

首先要把每个人手里分好的牌按照花色整合到一起，然后再对整合好的每一摞牌进行检查，最后分别输出检查结果。

我们具体拿一些数据来分析一下这个过程。

input是我们要处理的数据，这是一个文本文档。我们把它送到MapReduce模块里边，需要得到如右边所示的结果：统计出来每个单词出现的词数。

MapReduce怎么做呢？

既然是集群模式，肯定要把任务分配给不同的服务器。假设我现在把文件分成3块，就会在集群当中启动3个Map进程。每个进程拿到一块数据，接下来它会怎么做呢？

——正确答案：Map要对拿到的数据做一个处理，然后每个Map把处理结果交给Reduce，让Reduce去做一个统计。所以，Map先会将一行数据按空格键切分，变成若干个单词，然后单词每出现一次，计数为1，用一个键值对装起来，如下图：

在数据量很多的情况下，我们一般会让Map做单词的统计。

因为如果不做单词统计的话，它就会把这一堆<hello,1>，<world,1>直接交给Reduce，所有Map的统计结果都要通过网络传输给特定的Reduce进程的服务器，这样会造成不同服务器之间数据传输量很大。

所以Map会把每一个单词出现了多少次统计出来，比如hello出现了1次，world出现了2次。最后每个Map将手头的任务计算完毕就把结果存储起来，接下来就轮到Reduce上场。

在Map存储结果的过程中，Reduce会被启动，同时去拿Map处理的结果，当有多个Reduce的时候，Reduce不会把全部的结果都读取回来，而是只读取属于自己的分区，即分区Partition。

什么叫做分区呢？

分区就是对Map计算的结果做的一个分类，有多少Reduce，我就把结果分成多少类，也就是多少个分区。这是通过一个算法来完成的，这个算法是怎么回事，我们后面会做一个细节补充。

我们先以4个Reduce为例（一般情况下Reduce的个数少于Map的个数），假设每一个不同的单词刚好对应一个不同的分区。那么Reduce拿到的数据就是上图所示的样子。Reduce整合多个Map的计算结果，并且是键值对，键是单词，值变成由每一个Map提交的计算结果组成的集合。

Reduce要做的就是把这些集合里面的值加起来，得到最后的结果，也就是在整个文档里面，每个单词出现了多少次。

MapReduce处理数据的三个细节

接下来我们要讨论的是：在客户端将任务提交给集群处理之前，需要先对文件进行分片，那么这个分片是根据什么来分的呢？后续又是怎样将文件处理的任务分配给Map进程的呢？

——正确答案：提交任务的PC是Hadoop集群的客户端，客户端在提交任务之前要先将待处理的文件进行分片（Split）。我们的文件存放在HDFS中，而HDFS中的数据是按照数据块的形式分散存放在各个服务器里面的，不懂没关系，上图：

MR框架默认将一个数据块（Block）作为一个分片，分片是一个逻辑的概念，每个Map处理一个分片。比如说，第一个Map处理第一个分片，这个分片是从文件的开始，也就是偏移量为0的位置读取到偏移量为128M的位置；第二个Map处理第二个分片，这个分片是从偏移量128M读到偏移量256M，每一个分片对应启动一个Map来对其进行处理。

Map是一个计算程序，计算逻辑由做大数据应用开发的人编写。我们是在自己的PC上编写代码，编好了以后就要把代码提交到集群上去。

提交给谁呢？提交给Yarn，这是一个资源管理模块，它会为我们的任务执行分配资源。Yarn根据我们要处理的文件分布节点启动相对应的Map程序，因此Map进程也是分散在集群当中，这样就达到了分布式并发执行的效果。

至此为止，问题又来了：我们划分文件的目的是什么？

——正确答案：第一，做分布式运算，让计算变得更加高效，众人拾柴火焰高；第二，由于文件分散在集群当中，如果只有一个节点计算数据的话，就必须把数据集合到这个节点，这样做网络代价会很大。

最后，我们来把这个过程的细节问题补充完整。

第一个细节，就是文件被处理之前，也就是在启动MapReduce之前，必须确保待处理的文件放在HDFS上面。不然的话，计算节点没有数据，分布式计算就没法执行。

第二个细节，Map的输出先放入一个环形内存缓冲区，当缓冲区数据溢出时，需将缓冲区中的数据写入到本地磁盘，写入本地磁盘之前通常需要做如下处理：

1.分区（Partition）：默认采用Hash算法进行分区。现在我记录了hello，world等单词，对每一个单词求hash值，然后用这个hash值除以Reduce的个数并取余，Reduce的个数就等于分区的个数，每个Reduce对应一个分区。得到的余数就是最终要把这个单词放到哪一个分区。Hash算法的特点就是相同的对象求出来的值是一样的，所以相同的单词最终被送到相同的Reduce来处理。

这个过程稍微有一点点不好理解，没关系，一言不合就上图！

2.排序（Sort）：将Map输出的记录排序，例如将(’Hi’,’1’),(‘Hello’,’1’)重新排序为(‘Hello’,’1’), (’Hi’,’1’)。

3.组合(Combine)：这个动作MR框架默认是可选的，数据量很大的时候建议大家去做。这就是每个Map要完成的一个统计过程，例如将(’Hi’,’1’), (’Hi’,’1’),(‘Hello’,’1’), (Hello’,’1’)进行合并操作为(’Hi’,’2’), (‘Hello’,’2’)。

4.合并(Spill)：Map在处理后会产生很多的溢出文件(spill file)，这时需将多个溢出文件进行合并处理，生成一个经过分区和排序的SpillFile（MOF：MapOutFile）。为减少写入磁盘的数据量，MR支持对MOF进行压缩后再写入。至此，Map的任务就算是完成啦。上图，总结一下：

第三个细节，通常在Map 任务完成MOF输出进度到3%时启动Reduce，Reduce会去读取Map的输出文件MOF里面属于自己的分区，把这部分数据copy到自己的内存当中来。每一个Reduce会对应一个分区，然后去做后续处理，这是一个copy的过程。

当Reduce接收的数据量不大时，则直接存放在内存缓冲区中，随着缓冲区接收到的文件增多，Reduce要对接收到的MOF文件进行一个排序和合并，生成更大的文件，这是一个sort和merge的过程。

copy和sort、merge的过程我们一般统称为shuffle过程，这个过程会产生许多中间文件，最后一次合并的结果直接输出到用户自定义的Reduce函数。Reduce对合并后的文件进行处理，计算出我们需要的结果。

那么现在大家想想，为什么需要shuffle的过程呢？

——正确答案：大家考虑一下一个分区有好几个不同单词的情况，而且每一个Map都会输出一部分结果到同一个Reduce上，如果没有shffle的过程场面该有多么的混乱。

最后，完整的过程图双手奉上，大家要把每一个步骤看仔细。 

这就是MapReduce的一个完整的流程，其实很简单，只是有很多细节地方需要注意一下。

看完文章你是否也跃跃欲试，现在1元即可体验华为云的MapReduce，还能参与抽奖，不容错过。

备注：本文部分内容转载自微信公众号“叶子的浅浅时光”的《解读MapReduce》。