基于MapReduce框架所设计的计算是分布式计算。在Hadoop中,分布式文件系统为各种分布式计算需求服务。分布式文件系统就是增加了分布式的文件系统,将定义推广到类似的分布式计算上,可以将其视为增加了分布式支持的计算函数。从计算的角度上看,MapReduce框架接受各种格式的键值对文件作为输入,读取并计算后,最终生成自定义格式的输出文件。而从分布式的角度来看,分布式计算的输入文件往往规模巨大,并且分布在多个机器上,单机计算完全不支撑并且效率低下,因此Map Reduce框架需要提供一套机制,将计算扩展到无限规模的机器集群上进行。

在MapReduce框架中,将每一次计算请求称为作业。作业可分为两步骤完成。首先是将其拆分成若干个Map任务,分配到不同的机器上去执行,每一个Map任务将输入文件的一部分作为自己的输入,经过一些计算,生成某种格式的中间文件,这种格式与最终所需的文件格式完全一致,但是仅仅包含一部分数据。因此,等到所有Map任务完成后,进入下一个步骤,用以合并这些中间文件获得最后的输出文件。此时,系统会生成若干个Reduce任务,同样也是分配到不同的机器去执行,其目标就是将若干个Map任务生成的中间文件汇总到最后的输出文件中,这就是Reduce任务的价值所在。经过如上步骤后,作业完成,所需要的目标文件生成。算法的关键就在于增加了一个中间文件生成的流程,大大提高了灵活性,使其分布式扩展性得到了保证。

Hadoop术语的解释如表2-1所示。

表2-1　术语解释

(一)基本架构

在Hadoop架构中,作业服务器成为Master。作业服务器负责管理运行在此框架下的所有作业,也是为各个作业分配任务的核心。与HDFS的主控服务器类似,简化了负责的同步流程。执行用户定义操作的是任务服务器,每一个作业被拆分成多个任务,包括Map任务和Reduce任务等。任务是执行的基本单元,它们都需要分配到合适任务服务器上去执行,任务服务器一边执行一边向作业服务器汇报各个任务的状态,以此来帮助作业服务器了解作业执行的整体情况,以及分配新的任务等。

除了作业的管理者与执行者之外,还需要一个任务的提交者,这就是客户端。与分布式文件系统一样,用户需要自定义好所需要的内容,经由客户端相关的代码,将作业及其相关内容和配置提交到作业服务器,并随时监控执行的状况。

与分布式文件系统相比,MapReduce框架还有一个特点就是可定制性强。文件系统中很多的算法都很固定和直观,不会由于所存储的内容不同而有太多的变化。而作为通用的计算框架,需要面对的问题则更复杂。在不同的问题、不同的输入、不同的需求之间,很难存在一种通用的机制。对于MapReduce框架而言,一方面要尽可能抽取出公共需求,并实现它；另一方面要提供良好的可扩展机制,满足用户自定义各种算法的需求。

(二)计算流程

一个作业的计算流程如图2-5所示。

图2-5　作业执行过程

在每个任务的执行中,又包含输入的准备、算法的执行、输出的生成三个子步骤。按照这个流程,可以很清晰地了解整个MapReduce框架下作业的执行过程。

1.作业提交

一个作业在提交之前需要完成所有配置,因为一旦提交到了作业服务器上,就进入了完全自动化的流程。用户除了观望,最多也就能起一个监督作用。用户在提交代码阶段,需要做的主要工作是写好所有自定的代码,主要有Map和Reduce的执行代码。

2.Map任务的分配

当一个作业提交到了作业服务器上,作业服务器将生成若干个Map任务,每一个Map任务负责将一部分的输入转换成格式与最终格式相同的中间文件。通常一个作业的输入都是基于分布式文件系统的文件,而对于一个Map任务而言,它的输入往往是输入文件的一个数据块,或者是数据块的一部分,但通常不跨数据块。因为一旦跨了数据块,就可能涉及多个服务器,带来不必要的麻烦。

当一个作业从客户端提交到了作业服务器上,作业服务器将作业拆分成若干个Map任务后,会预先挂在作业服务器上的任务服务器拓扑树上。这是依照分布式文件数据块的位置来划分的,比如一个Map任务需要用某个数据块,这个数据块有三份备份,那么在这三台服务器上都会挂上此任务,可以视为一个预分配。

任务分配是一个重要的环节,任务分配就是将合适的作业分配到合适的服务器上。主要分为两个步骤。

(1)选择作业,然后是在此作业中选择任务。与所有分配工作一样,任务分配也是一个复杂的工作。不当的任务分配,可能会导致网络流量增加、某些任务服务器负载过重、效率下降等。不仅如此,任务分配还无一致模式,不同的业务背景,可能需要不同的分配算法才能满足需求。当一个任务服务器工作得游刃有余,期待获得新的任务的时候,将按照各个作业的优先级,从最高优先级的作业开始分配。每分配一个,还会为其留出余量,以备不时之需。举一个例子：系统目前有优先级3、2、1的三个作业,每个作业都有一个可分配的Map任务,一个任务服务器来申请新的任务,它还有能力承载3个任务的执行,将先从优先级为3的作业上取一个任务分配给它,然后再留出一个任务的余量。此时,系统只能再将优先级为2作业的任务分配给此服务器,而不能分配优先级1的任务。这样的策略,基本思路就是一切为优先级高的作业服务。(www.daowen.com)

(2)确定了从哪个作业提取任务后,具体的分配算法很简单,就是尽全力为此服务器分配任务,也就是说,只要还有可分配的任务,就一定会分给它,而不考虑后来的服务器。作业服务器会从离它最近的服务器开始,看上面是否还挂着未分配的任务(预分配上的),从近到远,如果所有的任务都分配了,那么看有没有开启多次执行,如果开启,考虑把未完成的任务再分配一次。

对于作业服务器来说,把一个任务分配出去了,并不意味着作业服务器工作完成,对此任务可以不管不顾了。因为任务可以在任务服务器上执行失败,可能执行缓慢,这都需要作业服务器帮助它们再来执行一次。

3.Map任务的执行

与HDFS类似,任务服务器是通过心跳消息向作业服务器汇报此时各个任务执行的状况,并向作业服务器申请新的任务。在实现过程中,使用池化的方式。有若干个固定的槽,如果槽没有满,那么就启动新的子进程,否则,就寻找空闲的进程。如果是同任务的直接放进去,否则杀死这个进程,用一个新的进程代替。每一个进程都位于单独线程中。但是从实现上看,这个机制好像没有部署,子进程是死循环等待,而不会阻塞在父进程的相关线程上。父线程的变量一直都没有调整,一旦分配,始终都处在繁忙的状况。

4.Reduce任务的分配与执行

Reduce的分配比Map任务简单,基本上是所有Map任务完成了,如果有空闲的任务服务器,就给分配一个任务。因为Map任务的结果星罗棋布,且变化多端,真要搞一个全局优化的算法,得不偿失。而Reduce任务执行进程的构造和分配流程,与Map基本一致。但Reduce任务与Map任务的最大不同是Map任务的文件都存于本地,而Reduce任务需要到处采集。这个流程是作业服务器经由此Reduce任务所处的任务服务器,告诉Reduce任务正在执行的进程,它需要的Map任务执行过的服务器地址,此Reduce任务服务器会与原Map任务服务器联系,通过FTP服务下载。这个隐含的直接数据联系,就是执行Reduce任务与执行Map任务最大的不同了。

5.作业的完成

当所有Reduce任务都完成之后,所需数据都写到了分布式文件系统上,整个作业才正式完成了。

(三)MapReduce程序的执行过程

基于MapRleduce算法编写的MapReduce程序的分布式执行过程如图2-6所示。

图2-6　MapReduce程序的执行过程

用户程序中的MapReduce类库首先将输入文档分割成大小为16MB～64MB的文件片段,用户也可以通过设置参数对大小进行控制。随后,集群中的多个服务器开始执行多个用户程序的副本。

由Master负责分配任务,如果总计分配M个Map任务和R个Reduce任务。分配的原则是Master选择空闲的Worker并为其分配一个Map任务或一个Reduce任务。

被分配到Map任务的Worker读取对应文件片段,从输入数据中解析出键值对,并将其传递给用户定义的Map函数。由Map函数产生的键值对被存储在内存中。

缓存的键值对被周期性写入本地磁盘,并被分成R个区域。这些缓存数据在本地磁盘上的地址被传递回Master,由Master再将这些地址送到负责Reduce任务的Master。

当负责Reduce任务的Master得到关于上述地址的通知时,它使用远程过程调用从本地磁盘读取缓冲数据。随后,Worker将所有读取的数据按键排序,使得具有相同键的对排在一起。

对于每一个唯一的键,负责Reduce任务的Worker将对应的数据集传递给用户定义的Reduce函数。这个Reduce函数的输出被作为Reduce分区的结果添加到最终的输出档中。

当所有的Map任务和Reduce任务都完成时,Master唤醒用户程序。此时,用户程序的MapReduce调用向用户的代码返回结果。

Map Reduce模型通过将数据集的大规模操作分发给网络上的各节点实现可靠性,每个节点将完成的工作和状态更新周期性地报告。如果一个节点保持沉默超过一个预设的时限,主节点记录下这个节点状态为死亡,并把分配给这个节点的数据发到别的节点。每个操作使用原子操作以确保不会发生并行线程间的冲突,当文件被改名的时候,为了避免副作用,系统将它们复制到任务名以外的另一个名字上去。

由于化简操作并行能力较差,主节点会尽量把化简操作调度在一个节点上,或者离需要操作的数据尽可能近的节点上。这种做法适用于具有足够的带宽、内部网络没有那么多的机器情况下的需求。