浙江大学计算机科学与技术学院 硕士学位论文 基于MapReduce模型的并行计算平台的设计与实现 姓名：万至臻 申请学位级别：硕士 专业：软件工程 指导教师：陈刚;寿黎但 20080501 浙江大学硕士学位论文摘要摘要随着互联网的迅猛发展，每天由网络产生的数据量越来越庞大。互联网企业面对这些浩繁的数据，常常陷入“数据丰富，信息贫乏”的尴尬境地。设计一个通用可扩展的平台，来有效地处理海量数据，不断地从中挖掘出对运营商有价值的信息，成为互联网企业发展的必然选择。MapReduce是由Google公司首先提出的，一种能在大型计算机集群上并发地处理海量数据的框架模型。使用者通过指定一个map函数将输入数据转化成为一系列中间键．值对，然后由一个自定义的reduce函数将具有相同键的值聚集起来，将结果输出。很多现实世界对海量数据的处理，都可以用这种模型来表示。本文在分析MapReduce模型的基础上，结合自身的特点，提出了一种并发处理海量数据的更通用、更可扩展的平台。首先，我们提出了海量数据并发处理平台的体系结构。该结构为客户端．任务调度与执行层．数据存储层三层架构。在客户端，通过可配置的XML文档提交用户任务。在进行任务调度与执行层设计时，我们首先提出了几点关键的策略，如通用平台策略、负载均衡策略、中间结果处理策略和容错策略。接着，我们提出了主控节点．分派节点．服务节点的三点式架构。其中，主控节点负责收集与处理其他节点的各种信息；分派节点负责解析、分派任务，获取任务执行结果；服务节点负责任务的具体执行。三种节点互相配合，共同完成数据的并发处理。接着，我们设计了存储海量数据的分布式文件系统。分布式文件系统具有优异的性能和吞吐率，较高的稳定性和良好的可扩展性。最后，我们在已经搭建好的平台上，进行了若干测试系统性能的实验。我们通过单机与并发执行用户任务的对比实验，说明了并行计算平台的高效。我们通过执行不同类型的任务，说明了如何在集群规模一定的情况下，实现性能调优。关键词：海量数据，映射规约，并行计算，分布式文件系统，负载均衡，容错机制，集群 浙江大学硕士学位论文AbstractAbstractAlong、析tlltherapiddevelopmentofinternet,thedataproducedbywebincreasesheavily．Itisallembarrassingpositionforwebsiteoperatorstofacea"richdata,littleknowledge'’situationinfrontofmassivedataset．Itisamust-choicetodesignacommonandextensibleplatformwhichhandlesmassivedataeffectivelyforwebsiteoperators．Thusitispossibleforthemtominethepotentialknowledge．MapReduceisasimpleandflexibleparallelprogrammingmodelproposedbyGoogleforlargescaledataprocessinginadistributedcomputingenvironment．Usersspecifyamapfunctionthatprocessesakey／valuepairtogenerateasetofintermediatevaluesassociatedwiththesameintermediatekey．Manyrealworldtasksaleexpressibleinthismodel．BasedontheanalysisofGoogle’SMapReduceandaccordingtoourownfeatures，weexploretheplatformofmassdatamanagementwhichismorecommonandextensible．Atfirst,weproposetheframeworkarchitectureofmassdataparallelprocessingwhichisaclient-schedulerandprocessor-datastoragestructure．Atclientend，userscommittaskbyconfigurableXMLdocument．Duringthedesignforlayeroftaskschedulingandexecuting，weproposesomekeystrategies，whichiscommonplatformstrategy,loadbalancingstrategy,intermediateresultsmanagementstrategyandfault-tolerantstrategy．Followingthesestrategies，weadoptmaster-dispatch-serviceframework．Themasternodeisresponsibleforcollectingstatusofeachnode；thedispatchnodeistodividethetasksettotaskunitsanddispatchthemtoservicenodes．atlastitwouldgettheresults；theservicenodestakeonactualdataprocessing．Next，wedesignadistributedfilesystemtohandlemassdatastorage．Thatsystemownsfineperformanceandthroughput，Strongstabilityandrobustness．’Finally,wecarryoutruntimeperformancetestsontheplatform．Comparingtheoutcomesofstand—aloneandparal．1elizationdataprocessing，wecandrawthe 浙江大学硕士学位论文conclusionthatparallelizationprocessingismoreeffective．Byexecutionsofdifferenttasks，weexplorehowtoachievetheoptimalperformanceunderthelimitofclusterscale．Keywords：massdata,MapReduce，parallelcomputing，distributedfilesystem，loadbalancing，fault-tolerance，cluster 浙江大学硕士学位论文图目录图目录图1．1Google典型集群⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯一6图1．2论文组织结构⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯8图2．1MapReduce执行流程⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯．．12图2．2基于MapReduce的程序在Google源代码树上的增长趋势⋯⋯⋯⋯⋯⋯⋯．16图2．3HDFS结构⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯．20图2．4Map／reduce在Hadoop上的实现流程⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯。2l图2．5应用框架的任务控制流图⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯22图3．1博客产品uV增长趋势图⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯．．25图3．2任务粒度与线程池实现动态均衡⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯28图3．3经过Combiner局部规约后的执行流⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯30图3．4并行计算平台总体框架⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯33图3．5三点式架构⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯。35图3．6各节点交互序列图⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯34图3．7主控节点(Master)类图⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯。37图3．8分派节点(DN)类图⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯38图3．9服务节点(SN)类图⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯40图4．1分布式文件系统硬件拓扑⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯42图4．2分布式文件系统架构⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯42图4．3文档ID(DoclD)与目录的映射算法⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯．43图4．4DFS工作流程图⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯一44图4．5文件读取流程⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯．46图4．6文件替换流程⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯一47图4．7文件上传流程⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯．．48图4．8文件删除流程⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯一49图5．1单机处理与并行处理性能比较⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯。53图5．2M印处理单元串行处理文件数与系统性能⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯．54图5．3排序运算执行时间随Reduce个数变化图⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯．．55图5．4(a)输入过程的数据传输速率⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯56图5．4(b)从M印到Reduce过程的传输速率⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯．57图5．4(c)结果输出过程的数据传输速率⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯57111 浙江大学硕士学位论文表目录表目录表5．1集群服务器配置⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯．51表5．2访问日志格式说明⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯⋯．．52IV 浙江大学研究生学位论文独创性声明本人声明所呈交的学位论文是本人在导师指导下进行的研究工作及取得的研究成果。除了文中特别加以标注和致谢的地方外，论文中不包含其他人已经发表或撰写过的研究成果，也不包含为获得逝望盘堂或其他教育机构的学位或证书而使用过的材料。与我一同工作的同志对本研究所做的任何贡献均己在论文中作了明确的说明并表示谢意。⋯一躲梦涛⋯～一洲日学位论文版权使用授权书本学位论文作者完全了解逝姿盘堂有权保留并向国家有关部门或机构送交本论文的复印件和磁盘，允许论文被查阅和借阅。本人授权逝姿叁鲎可以将学位论文的全部或部分内容编入有关数据库进行检索和传播，可以采用影印、缩印或扫描等复制手段保存、汇编学位论文。(保密的学位论文在解密后适用本授权书)学位论文作者签名：蓦乏跨签字日期：易。扩年否月‘日导师签名：签字日期：夕u舻年‘月G日 浙江大学硕士学位论文第2章MapReduce相关技术介绍对。2．3实现框架MapReduce接口可以有很多种不同的实现【11】【20】。应当根据不同的环境选择不同的实现。比如，适用于小型共享内存的实现，基于大型NUMA多处理器系统的实现041，还有基于大规模计算机集群的实现。下面是Google广泛使用的计算环境：用交换机网络连接的，由普通PC构成的超大集群1151。在这样的环境里：1)每个节点通常是双x86处理器，运行在Linux上，每台机器2-4GB内存；2)使用常用的网络设备。一般是百兆或千兆网络，一般情况下都用不到一半的网络带宽；3)一个集群中常常有成百上千台机器，所以，若干台机器的故障是难免的。4)存储时使用的廉价IDE硬盘，直接挂在每一个机器上。并且有一个分布式的文件系统【7】来管理这些分布在各个机器上的硬盘。文件系统通过复制的方法在不可靠的硬件上保证可用性和可靠性。5)用户向调度系统提交请求。每一个请求都包含一组任务，映射到这个计算机集群里的一组机器上执行。2．3．1执行概览Map操作是通过把输入自动分割成M个分区而分布到不同的机器上去执行的。输入可以在不同的节点上被并行地处理。而Reduce操作，则是通过把中间结果的键值空间切分成R块，来分布执行的，如可以使用切分函数肠嚣厅(铆)％足。切分函数与分区个数R可由用户指定。图2．1是MapReduce操作执行的流程图。12 浙江大学硕士学位论文第2章MapR饯iuce相关技术介绍⑤n’哗。lll赢_ll'嫩，’ll'嫩tl，Ⅱ“-'●，i铺l磋／一＼二·／睦J／splitl一÷。，一∥?sl岿it2～‘”＼-I／Ejsplit，澌l钐’≈～厂§lInl搬tfilesMal’IntermediatefilesReducepha辩{伽虹出disks)phase图2．1MapReduce执行流程oI嘲毗6k§当用户程序调用MapReduce函数时，引发了如下一连串动作(图2．1中的数字编号对应下列序号)：1)MapReduce库首先把输入文件切分成16到64兆的M块(可以通过参数调整)。接着在集群的不同机器上执行程序的拷贝。2)在所有进程中有一个比较特殊，它是主控程序master。其余的执行任务都是由主控程序分配的。主控程序分别分配了M个Map任务与R个Reduce任务。主控程序选择空闲的工作节点(worker)去执行这些任务。3)一个被委派执行Map操作的工作节点接受被切分的文件作输入，经过处理后，生成键一值对集合。集合被传递给用户定义的Map函数，Map函数产生的中间结果键．值对暂时被缓存起来。4)这些缓冲到内存的中间结果将被定时刷写到本地硬盘，这些数据通过分区函数分成R个区。这些中间结果在本地硬盘的位置信息将被发送回master，然后这个master负责把这些位置信息传送给reduce的工作节点。13 浙江大学硕士学位论文第2章MapReduce相关技术介绍5)当主控节点通知reduce工作节点中间结果的位置时，它通过远程调用从map的工作节点所在的本地硬盘上读取中间数据。当reduce的工作节点读取了所有的中间数据，他就使用中间结果的键进行排序[231，这样可以使得具有相同键的对都在一起。通常情况下，会有许多不同键的被映射到相同的reduce任务，所以，排序是必要的。如果中间结果集太大了，那么就需要使用外排序。6)Reduce节点根据每个唯一的中间键来遍历所有的排序后的中间数据，并且相关的中间结果集合传递给用户定义的reduce函数。对于本reduce区块，将输出到一个最终的输出文件。7)当所有的map任务和reduce任务都已经完成了的时候，主控程序激活用户程序。这时MapReduce返回到用户程序的调用点。当这些成功结束以后，MapReduce的结果数据存放在总计R个输出文件中(每个都是由reduce任务产生的，这些文件名是用户指定的)。通常，用户不需要合并这R个输出文件到一个文件，他们通常把这些文件作为输入传递到另一个MapReduce调用，或者用另一个分布式应用来处理这些文件，并且这些分布式应用能处理被分区的输入数据。2．3．2主控节点数据机构主控节点维护了一些数据结构。对于每个Map与Reduce任务，会在主控节点上记录下它们的状态(空闲，处理中或是已完成)，并且识别不同的节点。主控节点是由map任务产生的中间文件位置信息传递到reduce任务的管道。因此，对于每一个完成的map任务，主控节点保存此map任务产生的R个中间结果文件位置信息和大小信息。当主控节点接收到Map任务完成的消息后，它把中间文件的信息发送到到处于就绪状态的reduce任务的处理节点上。2．4容错机制由于MapReduce是设计用来支持成百甚至上千台机器并发处理的，所以它应14 浙江大学硕士学位论文第2章MapReduce相关技术介绍当拥有很好的容错机制。【12】【1sl[1912．4．1工作节点失效主控节点会定期ping每一个工作节点机器。如果在一定时间内没有工作节点的响应，主控节点就认为这个节点失效了。所有这台工作节点完成的map任务都被设置成为他们的初始空闲状态，并且因此可以被其他工作节点重新调度执行。类似的，所有这个机器上正在处理的map任务或者reduce任务都被设置成为空闲状态，可以被其他工作节点重新执行。在失效节点上的已经完成的map任务还需要再次重新执行，这是因为中间结果存放在这个失效的机器上，导致中间结果无法访问；已经完成的reduce任务无需再次执行，因为他们的结果已经保存在全局文件系统中了。当一个在节点A上执行的Map任务由于A节点失效，切换到B节点执行，所有执行Reduce操作的节点都被告知这一情况。这样的话，那些尚未从节点A上读取结果的Reduce节点要从B上获得数据。MapReduce可以有效地支持很大范围的节点失效的情况。比如，在一次MapReduce操作中，网络例行维护可能会导致大约有几十台机器在几分钟之内不能访问。MapReduce的主控节点简单的把这些不能访问的节点上的工作再执行一次，并且继续调度进程，最后完成MapReduce操作。2．4．2主控节点失效通常为主控节点数据结构设置周期性的检查点。这样当主控节点失效的时候，可以从最后一次检查点重新开始。不过，由于只有一个主控节点在运行，所以如果失效就比较麻烦。因此我们当前的实现上，如果主控节点失效了，就终止MapReduce执行。客户端可以检测这种失效并且根据需要重新尝试MapReduce操作。15 浙江大学硕士学位论文第2章MapReduce相关技术介绍2．5存储本地化在MapReduce计算环境中，网络带宽是相对稀缺的资源。可以通过让输入数据保存在构成集群机器的本地硬盘上的方式来减少网络带宽的开销【161[171。Google文件系统GFS把文件分成64M大小的块，在不同的机器上保存块的拷贝。MapReduce的主控节点有输入文件组的位置信息，并尝试在包含相应输入数据块的设备上分派map任务。如果不能分配，它就尝试分配map任务到尽量靠近这个任务的输入数据的机器上执行(比如，分配到一个和包含输入数据块在同一台交换机下的机器上执行)。当在一个足够大的集群上运行MapReduce操作的时候，大部分输入数据都是在本地机器读取的，这一过程并不占用网络带宽。2．6备用任务影响MapReduce操作总执行时间的原因之一是那些‘拖后腿’的节点：这些节点的任务执行得异常的慢。出现拖后腿的情况有很多原因。比如：一个机器的硬盘有点问题，经常需要反复读取纠错，然后读取输入数据的性能从30M／s降低到1M／s。集群调度系统已经在某台机器上调度了其他的任务，因为CPU／内存／本地硬盘／网络带宽等竞争的关系，导致执行MapReduce的代码性能比较慢。一个通用的机制‘捌可以用来减少拖后腿的情况。当MapReduce操作接近完成的时候，主控节点调度备用进程来执行那些余下的处理状态的任务。无论当最初的任务还是备用任务执行完成的时候，都把这个任务标记成为已经完成。通过调优了这个机制，通常只会占用多几个百分点的机器资源。但是发现这样做以后对于减少超大规模MapReduce操作的总处理时间来说非常有效。MapReduce的应用自2003年MapReduce类库在Google投入使用以来，它在很多问题领域都有广泛的应用【131。包括：1)大范围的机器学习问题；2)大范围的图形计算；3)索引系统重构；16 浙江大学硕士学位论文第2章MapReduce相关技术介绍4)网页内容提取；5)Google新闻面临的集群问题。图2．2为2003年至2007年间，基于MapReduce的程序在Google源代码树上的增长趋势【3】。一≯、／”／一．∥∥Jan4≥3磷彩粥Sep-03Jan4)4May4珥s{《“MJgn-0$May-05s|％埔5勘E珈∞赫嘭由S辍}pm6Jan-07May-07se04n'图2．2基于MapReduce的程序在Google源代码树上的增长趋势2．7开源MapReduce实现-Hadoop2．7．1概况Hadoop是大名鼎鼎的Lucene旗下的子项目，它原先是Nutch项目的组成部分，于2006年初从Nutch中分离出来成为一个独立的项目。Hadoop并非一个单纯用于存储的分布式文件系统，而是一个被设计用来在由普通硬件设备组成的大型集群上执行分布式应用的框架。Hadoop包含两个部分：一个分布式文件系统HDFS(HadoopDistributedFileSystem)，和一个MapReduce实现。因此，Hadoop的目标是为开发分布式应用提供一个框架，而不是像OpenAFS，Coda那样为存’储提供一个分布式文件系统。搜索引擎就是一种典型的分布式程序，Nutch就是基于Hadoop开发的。17跚雠撇撇瓣。谨稔782 浙江大学硕士学位论文第2章MapRe．ducc相关技术介绍2．7．2HDFSHadoop[141分布式文件系统(Hadoop’SDistributedFileSystem)受：Google文件系统(GFS)的启发，是建立在大型集群上可靠存储大数据集的文件系统。HDFS具有如下的特点：1)鲁棒性由于HDFS是部署在大量低成本硬件上的，那么硬件的故障是很正常的。整个HDFS系统将由数百或数千个存储着文件数据片断的服务器组成。其间经常会有一部分组件出现故障或是无法访问，这就意味着HDFS里的一些组成部分总是失效的。因此，故障的检测与快速恢复是HDFS一个很核心的设计目标。三种常见的故障是名字节点失效、数据节点失效和网络断开。每个数据节点周期性发送心跳消息到名字节点。网络断开会造成一些数据节点和名字节点失去联系，名字节点发现这种情况的根据是没有了心跳消息。名字节点标记这些数据节点已死，就不再将新的IO请求转发到这些数据节点上。而这些节点上的数据将对HDFS不再可用。这将导致一些块的副本数降低到临界值。名字节点检查所有的需要复制的块，并开始复制它们到其他的数据节点上。引发重新复制的原因还有：副本被破坏，数据节点上的磁盘损坏或增加了文件的副本数。当文件块从数据节点读出的时候，有可能被损坏。引发这种情况的原因可能是存储设备故障，网络故障或软件缺陷。HDFS客户端软件实现了HDFS文件的内容校验。当一个客户端创建一个HDFS文件，它为每一个文件块计算一个校验码并存储校验码在同一个HDFS名字空间中的一个单独的隐藏文件中。当客户端取这个文件内容时，它再根据这个校验码来验证从数据节点接受到的数据。如果不对，客户端可以从另外一个有该块副本的数据节点重新获取。2)数据组织HDFS是设计成支持大文件的。程序也是和HDFS一样地处理大数据集。这些程序写数据仅一次，读数据一次或多次，这就需要一个比较好的流读取速度。HDFS支持对文件的一写多读模式。HDFS典型的块大小是64M。18 浙江大学硕士学位论文第2章MapReduce相关技术介绍当一个客户端请求创建一个文件的时候，并不是立即向名字节点发请求。事实是，HDFS客户端在本地缓存文件数据，应用程序将写操作透明地重定向到临时本地文件。当本地文件堆积到大于HDFS块大小的时候，客户端联系名字节点。名字节点将文件名插入到文件系统当中，然后构造一个数据块。名字节点返回给客户端的响应包括数据节点的标识和目标数据块号，客户端再将本地的临时文件刷新到指定位置。当文件要关闭时，在把未刷入的数据传送完毕后，客户端通知名字节点此文件已经关闭。此时，名字节点提交文件的创建操作到持久化存储。假如名字节点在文件关闭之前挂掉，文件就丢掉了。上面的方式在仔细地考虑运行在HDFS之上的目标程序之后被采用。应用程序需要流式地写文件。如果客户端直接写到远程文件系统，而没有本地的缓冲，会对网速和网络吞吐量产生显著的影响。这种方式也不是没有先例，早期的分布式文件系统，如AFS也用客户端的缓冲来提高性能。3)简单一致模型大部分基于HDFS的程序对文件遵循的是“一次写入，多次读取”。一个文件一旦创建、写入、关闭之后就不需要修改了。这个假定简单化了数据一致的问题，实现了高吞吐数据访问。MapReduce程序或者网络爬虫程序都是非常完美地适合这个模型。Hadoop计划在将来实现文件的附加写入。4)移动计算比移动数据更经济在要被计算的数据所存储的位置来进行计算会提高效率，尤其是当数据集特别巨大的时候。这样消除了网络的拥堵，提高了系统的整体吞吐量。这个思想就是将计算迁移到离数据更近比将文件移动到程序运行的位置会更好。HDFS提供了接口，来让程序将自己移动到离数据存储的位置更近。5)名字节点和数据节点HDFS是主／从的体系结构。它拥有单个名字节点(NameNode)，这个一个主控节点，它管理者整个文件系统的命名空间和客户端对文件的接口。此外，还有大量的数据节点(DataNode)，负责存储实际的数据。HDFS暴露文件命名空间，允许用户数据存储成文件。在内部实现上，一个文件被切分为一个或多个块，这些19 浙江大学硕士学位论文第2章MapReduce相关技术介绍块被存储在一系列数据节点上。名字节点在名字空间内可以执行一系列文件操作如打开、关闭和重命名，它还决定了文件块对数据节点的映射关系。数据节点负责处理来自文件系统客户端的读写请求，同时它还可以执行块的创建、删除和复制操作。集群中单个名字节点极大地简单化了系统的结构。名字节点是仲裁者和所有HDFS的元数据的仓库，而用户的实际数据不经过名字节点。5)数据复制数据文件是以大小相同的块为单位存储在数据节点集群上的，为保证容错性，每一块都有副本。块大小与副本参数都是可配置的。程序可以指定文件副本的个数。名字节点负责所有文件块的复制。它会周期性地从数据节点接收心跳消息(Heartbeat)与块报告(Blockreport)。其中，心跳消息表明数据节点在正常工作，块报告包含了节点内所有块的列表。。大型HDFS实例所依赖的集群往往分布在不同的机架上。不同机架上节点的通讯需要通过交换机，而一般情况下，同一机架机器间的网络带宽要大于不同机架间。在启动阶段，每个数据节点会向名字节点注册其所在的机架号。一种简单的复制策略就是将副本放在不同的机架上，这样不但能有效防止整个机架上的节点失效，还能在读数据时提高吞吐量。这种将副本均匀分布的策略可以保证在某些节点失效的时候轻松地实现负载均衡，当然，它的缺点就是增加了写操作的代价，因为一次写操作要向不同机架上的块传送数据。一般情况下HDFS的复制策略是这样的：当副本数为3时，将副本1放置在本机架的一个数据节点上，副本2放置在本机架的另一个节点上，副本3放在不同机架的节点上。这种方式减少了机架内的写流量，提高了写的性能。机架失效的几率远小于节点的失效；这种方式不会影响数据的可靠性和可用性。但是它减少了读操作的网络聚合带宽，因为文件块存在两个不同的机架，而不是三个。这种机制在提高写性能的同时，并不以牺牲数据可靠性和读性能为代价。如图2．3是HDFS的体系架构。 浙江大学硕士学位论文第2章MapReduce相关技术介绍2．7．3HadoopMapReduce图2．3HDFS结构map／reduce操作在Hadoop中的实现流程见图2．4。21 浙江大学硕士学位论文第2章MapReduce相关技术介绍图2．4Map／reduce在Hadoop上的实现流程如图2．4所示，一个map-reduce任务将输入数据集分割成相互独立的若干块，以便map操作能以完全并发的方式进行【25】【261。要指出的是，分割不需了解文件的内部逻辑结构，具体的分割模式既可以自己指定，也可以使用Hadoop已定义的几种简单分隔。当单个map任务开始时，InputFormat类会分析输入的文件，产生对。这时，用户自定义的mapper类可以对键值对进行任意的操作。完成后，则调用OutputCollect类中的方法重新收集自己定义的键值对。此时键与值的类型不必与输入时的相同。产生的输出必需用一个Key类和一个value类，这是因为Map的输出结果要被以SequenceFile的形式写入磁盘。map的输入和输出不必在类型上有联系。·对于每个map输出的结果，可以通过combiner进行初步的合并。合并工作是在进行map操作的同一个节点上进行的。Combiner的具体操作与reducer相同， 塑兰奎兰堡主堂篁笙茎墨!兰坚翌坠!!塑鲞垫查坌塑它的主要目的是减少中间结果传输时的网络流量【2711291。在中间结果传送给用户自定义reducer操作之前，还需要对中间结果进行分区。这个工作由patitioner类来完成，默认是以HashPartitioner类用key类的哈希函数产生的哈希值来区分。然后将每个map中间文件中具有相同计算结果的键值对合并到同一个文件，这样就可以保证同样的key只会被送给同一个reducer来处理。当一个reduce任务开始时，它的输入是分散在各个节点上的map的输出文件里。这就需要reduce通过HTTP将中间结果的相应分区取到本地。与此同时，系统会将Reducer的输入根据key来排序，使得相同的key排列在一起。接着，通过用户自定义的reducer函数，对具有相同key的键值对进行归并。Reduce的结果通过OutputCollector类输出到HDFS。在整个运行过程中，框架使用Reporter类来报告进度信息，设置应用级别的状态消息。如图2．5所示，应用程序框架包含唯一一个主控JobTracker，在每个节点上，／一，一，√，一，／’J＼．图2．5应用框架的任务控制流图＼酵逐JL卜非|I 浙江大学硕士学位论文第2章M印Redu∞相关技术介绍拥有一个TaskTracker，在任务执行不同的阶段，TaskTracker监控不同的Task。JobClient负责向系统提交用户定义Mapper，Reducer等类，设定各种系统参数，将工作任务提交给JobTracker。JobTracker接受任务后，负责在工作节点调度任务并且周期性地监控它们，处理失败的任务。而TaskTracker则负责具体执行map／reduce操作并周期性地向JobTracker报告进程信息。2．8本章小结MapReduce的编程模式在Google成功应用于许多方面。这种成功应用归结为几个方面：首先，这个编程模式易于使用，即使程序员没有并行或者分布式系统经验，由于MapReduce封装了并行的细节和容错处理，本地化计算，负载均衡等等，所以，使得编程非常容易f30]D11。其次，大量不同的问题都可以简单通过MapReduce来解决。例如，MapReduce用于产生Google的web搜索服务所需要的数据，用来排序，用来数据挖掘，用于机器学习，以及很多其他系统。第三，Google已经在一个好几千台计算机的大型集群上开发实现了这个MapReduce。这个实现使得对于这些机器资源的利用非常简单，并且因此也适用于解决Google遇到的其他很多需要大量计算的问题。我们也从MapReduce上学会不少内容。首先，执行编程模式使得并行和分布式计算非常容易，并且也易于构造这样的容错计算环境。其次，网络带宽是系统的资源的瓶颈。我们系统的一系列优化都使因此针对减少网络传输量为目的的：本地优化使得我们读取数据时，是从本地磁盘读取的，并且写出单个中间数据文件到本地磁盘也节约了网络带宽。第三，冗余执行可以减少慢机器带来的影响，并且解决由于机器失效导致的数据丢失问题。24 浙江大学硕士学位论文第3章通用并行计算平台设计第3章通用并行计算平台设计本章介绍基于MapReduce的通用并行计算平台的设计。首先，我们会介绍此系统的产生背景；接着，谈谈在系统设计过程中若干关键的策略；然后，介绍系统的整体架构；最后，我们会详细介绍系统各子模块的结构。3．1系统背景网易杭州研究院数据分析与挖掘组担负着网易博客、相册与圈子产品日常的数据统计和分析工作，每天要处理数以亿计的日志记录和数据库记录。以往的做法是，将统计与分析的任务按照相似度进行划分，将它们分布在不同的服务器上并发的执行。这种方法的优点是简单易行，配置与实现都很方便。但是，随着网易博客等产品的发展，其用户群越来越庞大，访问量逐步增长。如图3．1所示，博客产品的uV，已经从上线伊始的60万左右，逐渐增长到磐j墨互曩，哆，霉'∥'《鬻黪，一‘≮≯。，。Z，。；¨。劳。j々。蒙鬻攀豢然嚣藜；j曩。纛。j¨¨缀一囊筠；誊‘甏≯爰缀爹薯叠i，《雾鬈—i．7I『 ≤；攀誓豢攀蒸溱爹i黉寨：{j；雾曩曩¨i：一一，；≯?i爹瀵童≤!誉?戮≯爹≤≯糍。j‘_：}；{}_≯oi受‘0≥鼍一jg童j磐j”jj，●i“_『j+iio⋯。?o誓■囊_i。_≯。o澎曩jI¨-一。≥i¨{⋯i10≥V曩≮≥j0li_。冀|_：|j“一ow。⋯～¨j?，?。『一≯07。jt7i。7√⋯。≤嘉簿joj『：，’。荔“l0兹誊雾1；l；荔爹。誊、‘_÷j，。_搿一磐‘：，≯_j0_≯毫⋯i_j。曩。∥?∞≤爹i藩j薹|i。-『jj≯一j÷?0ij_7“曩一。j纛j÷一≯一≥o≯≯。。I|。象“+l|j?；¨¨二i7i 浙江大学硕士学位论文第3章通用并行计算平台设计现在的330万。就访问日志的数据量而言，也已经翻了几番。在这样的情况下，简单依靠任务划分方式进行统计分析的弊端越来越显现，表现为：自动化程度低，对单个服务器性能依赖过高，任务调度和资源负载均衡随着统计分析规模的加大愈加困难，个别任务很难在单服务器下完成，服务器资源浪费严重。针对这些问题，我们需要开发这样一个框架：它可以在短时间内进行海量数据的处理；它在较短的或可接受的时间内，通过有效的调度和负载均衡，实现海量数据的并发处理；它应该提供良好的接口，让使用者丝毫不必关心繁杂的任务调度而专注于问题本身，如有可能，常用任务只需通过简单的配置即可进行；它应当为海量数据提供一个安全可靠的存储平台；此外，它还应当具有良好的可扩容性，可以随时增加计算资源而不必对框架进行任何修改；它应该很好的利用公司现有的分布式集群和文件系统，具有良好的容错性；它应当采用“切分数据，任务随数据分布"的策略，而不再是过去那种“切分任务，数据随任务分布”的策略；最后，它还应当坚持“框架与应用松耦合"的原则，提供完备的基础性服务，使系统不仅能很好的支持统计分析任务，还能很好的支持其他需要在海量数据上进行运算、分析和挖掘的工作，具有良好的可扩展性。Google提出的MapReduce计算模式，遵循了奥卡姆剃刀原N(Occam’sRazor)，即“如无必要，勿增实体”，用简洁的框架解决了海量数据上的计算瓶颈问题，对于我们解决面临的问题有很好的借鉴意义。然而，由于Google的MapReduce提供的是一种编程模式，它为了提高可控性，向用户提供编程接口进行并行计算，可控性有余，普适性不足。而我们的框架，由于更具有基础性和通用性，不妨在可控可测性和可扩展性之间做一个权衡，即，适当牺牲可控可测行，提高其普适性。下面我们开始介绍并行计算平台在设计中的若干策略。 浙江大学硕士学位论文第3章通用并行计算平台设计3．2若干关键的设计策略3．2．1通用平台策略与Google与Yahoo的MapReduce实现相比，我们的平台更通用，框架更可扩展。为了达到这一目标，我们采用了以下几种策略：1)客户端任务提交客户端任务的配置不是通过函数调用的方式，而是采用XML文档配置任务。在这种情况下，用户自定义任务不仅可以是已编译可运行的类，或是．jar包，还可以是一条shell命令，一个Bash脚本，甚至是一个工作目录。这种方式取消了对任务表达描述形式的限制，使框架更具通用性。2)对用户自定义任务的监控由于用户定义任务的组织方式有多样，我们没有也不可能对任务执行的过程进行严密的监控。只是在一个任务单元执行完后，根据其返回的状态码来判断任务是否顺利地完成。如成功，执行下一步操作；若失败，则重新调度执行失败的任务。3)任务串行任务操作分两种：Map与Reduce。但框架对这两种任务的执行逻辑，并无任何限制，所不同的仅是在输入输出文件上。即对M印任务的输入文件要进行切分，将不同输出文件的同一分区经过排序合并后，传给相应的Reduce操作。而Reduce的输出则作为结果文件。框架支持将多对Map／Reduce任务串行起来，形成I铂专巧--9．m2专吃专⋯专％一，；tJ的执行序列，完成更复杂的任务。在此种情况下，上一轮Reduce操作‘一。的结果不再作为结果输出，而是作为本轮Map操作他的输入。注意，为减少网络传输流量，咯一。与％位于同一节点上。3．2．2分布式负载均衡与任务调度策略系统的集群要求能够利用在公司内部使用的各类服务器上，这些服务器可能 浙江大学硕士学位论文第3章通用并行计算平台设计闲置，也可能运行着其他多种多样的应用程序。各台服务器的硬件配置和系统负载千差万别。要把各类复杂的服务器后台进行良好的整合和管理，使它们能够对外呈现一个整体和透明的分布式计算支撑平台，必须解决好分布式任务调度和负载均衡问题：我们通过如下方法实现负载均衡策略：1)任务智能调度如图3．2所示，我们根据任务的粒度来选择服务节点。做出选择判断的依据有二：一是任务的粒度，在任务的类型一致的情况下，影响粒度最主要的因素是输入文件的大小；二是服务节点的硬件资源尤其是CPU核个数。在具有多核处理器的节点上，线程池会适当地多执行一些并发的线程。因此，实际的map／reduce任务单元通常会多于服务节点的数目。比如，一个任务具有200000个map任务单元和5000个reduce任务单元，实际的服务节点数可能只有2000个。ProcessTime⋯⋯⋯⋯一一◆UserProgramMapReduce()wait⋯DNAssigntaskstoworkermachines⋯争Mapl“_Workerl卜：。鳞孑习Worker2艮篓、≯蒜妇。侉一警二琴鬻Worker3酬闲獭黼卜赫ucej霉IWorker4豳瓣瞩巍藤隧麟霾蘑一；眇∞-2。麦图3．2任务粒度与线程池实现动态均衡我们通过工作节点的心跳协议，实时收集各个服务节点的Load、CPU、内存、网络等负载信息，作为任务调度的选择基准，把各台服务器的负载控制在合理范围内。通过工作节点的任务汇报协议，实时监控任务的执行，并把任务工作状态向用户汇报。 浙江大学硕士学位论文第3章通用并行计算平台设计用户程序的代码可能会让map或者reduce函数在处理某些记录的时候crash掉。调度器需要根据任务是否可以容忍部分数据丢失来选择是否跳过损坏的数据。在计算能力冗余时，在数据复制节点上启动冗余任务，确保任务执行成功。2)网络均衡各个服务器节点间的通信只能通过以太网进行，实践中可以发现，网络带宽可能是整个系统最稀缺的资源。例如，单机处理5G的数据可能只需要10分钟，但是网络传输5G的数据可能耗费30分钟。因此，必须采取一系列优化措施，以减少网络传输量。我们采取了各种措施来尽可能减少跨网关的网络流量。比如数据文件本地化。当一个服务节点宕掉时，我们决定重新执行的节点时，尽量选择处于同一网关下的节点(或者是在同一机架上)，如此可以避免网关带宽的限制。3．2．3中间结果文件的处理策略通常Map阶段的中间结果被散布分区后，作为下一阶段任务的输入。默认的分区函数采用了Hash(key)％R的方法。当然，还可以指定其他分区函数，如，可以将中间结果的不同字段作为分区的目标字段，或者是经过处理的字段。在一些情况下，在那些经过Map操作处理过的中间结果中，会有很多重复的键存在。比如计算文章中单词频率的应用，中间结果难免会出现许多诸女H这样的键值对。如果将所有这些中间结果不做任何处理，就发送到Reduce操作所在的节点，无论对于网络的流量，还是Reduce操作的负担，都加重了。因此，如图3．3所示，我们允许用户自定义一个合并(combiner)方法，对中间结果进行局部的规约。Combiner操作在每个进行M印操作的节点上运行，通常情况下，它采用与Reduce操作同样的流程，唯一不同的是，系统采取不同的方式对待两类操作的输出文件。Reduce操作的结果被写到最终的输出文件，而Combiner操作的结果仍然被输出到中间文件进而发送给Reduce操作。 浙江大学硕士学位论文第3章通用并行计算平台设计图3．3经过Combiner局部规约后的执行流3．2．4容错机制中超时重发策略本框架的容错机制，主要采用超时复发策略，具体来说，就是：Master周期性的ping各个SN(这里我们不考虑DN，一来DN通常为单节点，二来其负载不大．)，检测他们的状态。当一段时间之后没有响应，master认为该SN已经出现故障。在该SN上正在处理的map或reduce任务被设置成空闲状态，以便重新调度。完成的map任务需要重新执行，那是因为它们的输出是存储在出现故障的本机磁盘上，而导致不可访问。完成的reduce任务输出结果是存储在DFS上，不存在这个问题。现在，我们通过一个具体的模型，从理论上研究这些策略的性能特点。模型如下：对master来说每个map任务需要时间吒+C册来完成，每一个reduce任务需 浙江大学硕士学位论文第3章通用并行计算平台设计要时间啡+C来完成。其中屯，6，分别指map任务和reduce任务在sN上执行的基本时间，他们都是常量。而q，C是服从参数分别为丸，4指数分布的随机变量，是完成一个任务的过程中不确定的时间。Master周期性ping每个SN，ping的周期为fo，以检测这些节点是否失效。在这里，我们选取map阶段的调度性能进行分析，对于reduce阶段的情况可以进行相应的修改得到。这里引入两个性能衡量指标，定义如下：定义1-如果一个节点在to时刻获得任务单元，在fl时刻将任务的结果提交，那么我们定义瓦坍蜘鲫珂=‘一乇，它是节点完成任务单元的总时间。这里不仅包括任务执行需要的时间，还包括网络延迟、调度延迟和等待时间。定义2：如果一个系统总共完成M个任务单元，耗时为T，那么定义单位时间内完成的任务单元数乃为％。下面对超时复发策略进行分析，首先引入超时时限T，做如下讨论：I)当to->T时：此时master检测各个SN是否有效的时刻在超时时限之后，在整个处理过程中，master的ping检测并没有起到提前结束失效节点的任务之作用。这里引入一个新的概率风，表示在T时间内，节点失效的概率。服务器以概率g判定一个客户端不会返回有效的结果，有：q=l-(1-Po)(1-e以仃训)式中，P一厶(卜％’是SN在可以返回结果时，返回的结果在T之前的概率。由此可以得到单个map任务完成时间即咒。舢。。的表达式如下：瓦。枷删树=(1一g)(k+C肼)+g(Z0，。M耐+丁)于是：咒删26m+q+9％一g)则有：3l 浙江大学硕士学位论文第3章通用并行计算平台设计乃2％(瓦～)2愿+炊+g％一g))II)当tol-dfsManagerI+parseTaskSetConfl90●山-tasklD+cleanAIIDoneTasks0MPRTa摹kSet+sondTask0+shutdownO+dispatchO4．t：aSkCleaner-MpRR^r●moTItk!≈^●-·name+isWorkDone()II—．step+requestMainSnUrl0Il—b●nPath4-rurtl,______‘·CUst}al>)--_---l_一广——哪-commandStr■■●●⋯●■■■■一●■■■■一—taskLIstMPRMapTaskSet-seleclM)r-reduceSnSelector：ServiceNodeSelector+dispatchTasks0．reduceldLIst．-mergeTaskUnltOutput0《《●nterface>>_—redUceIdLIstlndex：●nt+waitTasks0ServIceNodeSelectorl#mergeTaskUnitOutputO．．cleanAIITasks()+dlspatchTasks0二嚣菇；品名翁：菩bs捃fusDI图3．8分派节点(研、I)类图MPRDispatchNode请求调用的入口。负责开启与关闭DN，控制与Master的通讯。解析任务配置。ServiceNodeSelectorMPR服务节点挑选接口类。LocationSelector实现ServiceNodeSelector，负责服务节点的动态挑选。MPRTaskSet任务集合类，是任务单元的容器，负责整个任务队列的执行。MPRMapTaskSetM印任务集合类，负责Map任务集合。MPRReduceTaskSetReduce任务集合类，负责Reduce任务集合。MPRTaskUnit任务单元类，发送到服务节点上的任务的基本单位。3)服务节点服务节点是任务实际执行的所在。任务集合被划分为有次序的任务单元，初始的输入数据也被切分为大小相同的块，在每个服务节点上，任务单元运行在数据块的集合之上。如果任务单元并非是任务集合的最后一个，那么其输出文件会 浙江大学硕士学位论文第3章通用并行计算平台设计作为下一个任务单元的输入文件而被传入。另外，服务节点还要定期收集本机的负载信息，定期向master发送。其具体工作如下：a)调度本机M印任务处理输入／输出数据●读取输入数据：从分布式文件系统或是本机读取Map的输入数据。-调度执行Map任务：调度Map任务，对任务执行状态进行监控，并向Master汇报。●输出数据分区：把map的输出数据通过分区函数切分成多个子集作为reduce的输入，各子集之间数据不重复。b)调度本机Reduce任务处理输脯出数据一预排序功能：在把map的分区数据合并为reduce的输入前，可以选择对数据进行排序，包括外排序和内排序。一调度执行Reduce任务：调度Reduce任务，对任务执行状态进行监控，并向Master汇报。_输出数据处理：对reduce输出的数据，也可以根据需要进行排序去重等后期处理。c)调度Combine任务一在Map瓜educe之间可以加上合并流程，在Map的任务机上对输出数据进行再处理，如合并相同的key，以减少网络传输开销。如图3．9，为服务节点(ServiceNode)的类图。MPRServiceTask服务节点工作主类，负责设置任务执行的各种参数，读取输入文件，对输出文件进行处理，进而执行任务。MPRJi．ileDivider按照一定的切分模式，负责文件的切分。MPRFileMerger负责文件的合并，在合并过程中按照指定的字段进行排序。MPRMapTask继承MPRServiceTask，具体负责Map阶段的任务。MPRReduceTask继承MPRServiceTask，具体负责Reduce阶段的任务。MPRCombineTask继承MPRServiceTask，使用adapter模式，其中，adapter为MPRCombineTask，adaptee为MPRFileDivider与MPRFileMerger，target为39 浙江大学硕士学位论文第3章通用并行计算平台设计MPRServiceTask，在处理输入输出文件时，它将MPRFileDivider与MPRFileMerger包装为可以被调用的接口。rvICeTaSbsK¨)：IonIgbinF¨●cornmandStrattLachFlietaskTimeout：Ion口InPutListoutputListworkDirFileinPutDIrFIlelastCheckTime：longdfsManaaersetAtlr篁lchFiJesetBinF¨et)setCommandStrOsetDFsManag·rOsernmeout(’setlnputList‘lsetOutPutList0shutdown()getlnPutFile0handleOutputFIle0rUna8oaUC●TaSner口oroo卜¨enputlDList，artitionNum：intnerger：MPRFileMerger；oftgovSoquonc●：intnanaleAttrlDute(JhandtelnputFlie()setlnputList‘’3．5本章小结omnPutIDList>utputIDListnPutPartition：-nt>utputPartition：Int|ivider：MPRFlieDiv|dernerger：MPRFileMerger'artiotionKevSequenc●：inthandleOutputFiI《handlelnputFile0setOutputList0sotlnK)utListllutputlDListartitionNumivider：MPRFlieDividernWorkDirartiotionKovSeauence：intndleAndleOutputFile0tPartiotionKeySequence0touti)utLisHl<