Mapreduce介绍

Mapreduce1631703周泽霞•如果想统计下过去50年毕业生毕业论文出现最多的几个单词，看看大家都在研究些什么，那收集好论文后，该怎么办呢？•写一个小程序，把所有论文按顺序遍历一遍，统计每一个遇到的单词的出现次数；•写一个多线程程序，并发遍历论文，需要同步共享数据；•把作业交给多个计算机去完成：可以使用方法一的程序，部署到N台机器上去，然后把论文集分成N份，一台机器跑一个作业。这个方法跑得足够快，但是部署起来很麻烦，我们要人工把程序copy到别的机器，要人工把论文集分开，最痛苦的是还要把N个运行结果进行整合；情景MapReduce产生背景及概念Mapreduce运行系统实现Mapreduce优化研究目录MapReduce框架及编程模型MapReduce产生背景及概念大数据处理问题面临的挑战并行/分布式计算大数据的产生1423MapReduce产生背景大数据的产生1大数据处理问题•怎样统计一个文档集中每个word出现的次数?•怎样给一个文档集建立倒排索引？•怎样从一次crawled的web数据里构建出一个反向webgraph?•Tryonthiscollection:2006年初，我们在国内搜集了870Million不同网页,共约2TB.2并行/分布式计算•High-endMainFrame.vs.commodityPCCluster规模小可靠性高性价比高规模大But可靠性差3面临的挑战•大规模数据处理面临的困难•大规模PC机群scalingreliablyishard!•On1000sofnodes•MTBF(平均故障间隔时间)1day•Withsomanydisks,nodes,switchessomethingisalwaysbroken•并行/分布式程序开发，调试ishard!•数据如何划分•任务如何调度•任务之间的通信•错误处理，容错…RuntimeSystem良好可扩展性良好的容错能力ProgrammingModel一定的表达能力很好的简单易用性4MapReduce概念•MapReduce是一个分布式/并行编程模型，一个处理和生成超大数据集的算法模型的相关实现。——任务的分解与结果的汇总。•程序员只要按照这个框架的要求，设计map和reduce函数，分布式存储、节点调度、负载均衡、节点通讯、容错处理和故障恢复等操作都由mapreduce框架自动完成，设计的程序有很高的扩展性。InputsplitshuffleoutputMapreduce编程模型mapingshuflingreducingMapReduce实例——Wordcount•一批文件（规模为TB级或者PB级），如何统计这些文件中所有单词出现的次数？•方案：首先，分别统计每个文件中单词出现次数，然后累加不同文件中同一个单词出现次数。MapReduce编程模型—内部逻辑MapReduce将作业的整个运行过程分为两个阶段：和Map阶段由一定数量的MapTask组成输入数据格式解析：InputFormat输入数据处理：Mapper数据分组：PartitionerReduce阶段由一定数量的ReduceTask组成数据远程拷贝数据按照key排序数据处理：Reducer数据输出格式：OutputFormatMapReduce编程模型Datastore1Datastorenmap(key1,values...)(key2,values...)(key3,values...)map(key1,values...)(key2,values...)(key3,values...)Inputkey*valuepairsInputkey*valuepairs==Barrier==:Aggregatesintermediatevaluesbyoutputkeyreducereducereducekey1,intermediatevalueskey2,intermediatevalueskey3,intermediatevaluesfinalkey1valuesfinalkey2valuesfinalkey3values...•读入数据:（key，value）对的记录格式数据•Map:从每个记录里extractsomethingmap(in_key,in_value)-list(out_key,intermediate_value)处理inputkey/valuepair输出中间结果key/valuepairs•Shuffle:混排交换数据把相同key的中间结果汇集到相同节点上•Reduce:aggregate,summarize,filter,etc.reduce(out_key,list(intermediate_value))-list(out_value)归并某一个key的所有values，进行计算输出合并的计算结果(usuallyjustone)•输出结果Shuffle实现MapReduce编程模型——Wordcount•输入：onedocumentperrecord•用户实现mapfunction，输入为•key=documentURL•value=documentcontents•map输出(potentiallymany)key/valuepairs.•对document中每一个出现的词，输出一个记录word,“1”map(Stringkey,Stringvalue)://key:documentname//value:documentcontentsforeachwordwinvalue:EmitIntermediate(w,“1″);MapReduce编程模型——Wordcount•MapReduce运行系统(库)把所有相同key的记录收集到一起(shuffle/sort)•用户实现reducefunction对一个key对应的values计算•求和sum•Reduce输出key,sumreduce(Stringkey,Iteratorvalues)://key:aword//values:alistofcountsintresult=0;foreachvinvalues:result+=ParseInt(v);Emit(AsString(result));Mapreduce运行系统实现GoogleMapReduce架构SingleMasternodeManyworkerbeesManyworkerbeesMapReduce操作FinaloutputwrittenInitialdatasplitinto64MBblocksComputed,resultslocallystoredMasterinformedofresultlocationsMsendsdatalocationtoRworkersHadoopmapreduce架构JobTrackerMapReducejobsubmittedbyclientcomputerMasternodeTaskTrackerSlavenodeTaskinstanceTaskTrackerSlavenodeTaskinstanceTaskTrackerSlavenodeTaskinstance资源管理器：监测空闲资源，分配给TaskTracker；任务管理器：监控所有TaskTracker与job的健康状况，一旦发现失败，就将相应的任务转移到其他节点；TaskTracker会周期性地通过Heartbeat将本节点上资源的使用情况和任务的运行进度汇报给JobTracker，同时接JobTracker发送过来的命令并执行相应的操作HDFSMapreduce优化研究容错机制性能优化——数据倾斜MapReduce容错机制普通节点级•TaskTracker周期性（默认是3秒）的向JobTracker发送心跳信息，报告节点的健康状态和在节点上运行的任务的执行情况。•当JobTracker在预设的时间段内（默认是10分钟）没有收到来自某个TaskTracker发送的心跳信息时，JobTracker会以为其已经宕机，并将该TaskTracker上未完成的Map或者Reduce任务重新调度到其他的TaskTracker上再次执行。备份Google的研究报告表明，1-5%的硬盘会发生失效，服务器的失效率达到了2-4%。对于一个新的集群来说将会有超过1000次的节点会产生失效。任务级•一个任务(Map/Reduce)失败•调度器会将这个失败的任务分配到其他节点重新执行；•一个节点宕机•本机节点上已经完成运行的Map任务和正在运行中的Map和Reduce任务都将被调度重新执行，•在其他机器上正在运行的Reduce任务也将被重新执行，如果该Reduce任务所需要的Map的中间结果数据因为那台失效的机器而丢失了。再执行哪些任务需要再次调度执行？哪些任务需要再次调度执行？•已经失败的任务•可能失败的任务作业运行过程中一些执行的特别慢的任务，MapReduce认为其可能成为失效的任务；调度器会在其他的节点上重新调度这个任务执行。此时，一般会有两个相同的任务在同时执行，最后哪个任务先完成，那么这个任务就算完成了，而没有完成的那个任务就会被杀死。改进：LATE（LongestApproximateTimetoEnd）的调度策略。首先估算所有在运行任务的剩余完成时间（LATE），然后找到一个具有最长的剩余时间的任务，最后对这个任务进行推测执行存在的缺陷•在发生错误之前进行相应的备份工作；•在发现了失效功能模块之后进行的相应的恢复动作；•不能快速地及时的定位出失效的功能模块或者节点；如何快速地寻找和定位出失效的节点容错机制优化快速地寻找和定位出失效的节点心跳代表着节点的健康状况，反映节点是否还“活”着；反映节点上任务的运行情况：TaskTracker会时不时的通过心跳向JobTracker汇报在其上运行的任务的运行情况。自适应的超期时间基于信誉的探测模型基于信誉的探测模型，对信誉值进行调整设置自适应的超期时间估算整个作业的运行时间自适应的失效检测整体框架负责接收和发送心跳两段心跳：Hadoop集群的心跳默认为3秒，太固定和严格心跳不能太短导致过于频繁；不能太长影响作业的完成时间；自适应的超期时间•对于每台TaskTracker来说，默认的超期时间是10分钟，认为是优秀的值，主要的考虑的是它覆盖各种各样大小的作业，在平均情况下表现优异。针对的是存在超过几个小时的作业，也有只有几分钟的作业的集群。•一个短作业优先的集群中，10分钟的设置非常不合理。估算执行时间：map阶段shuffle阶段reduce阶段超期时间设置：上限=10基于信誉的探测模型•整个作业在reduce阶段执行时有一台节点宕机了，那么其他节点的reduce任务将出现数据接收的异常（fetch-error），这是因为宕机结点上的map任务的数据丢失了。Hadoop将在一个超期时间以后发现到这台宕机的节点，然后再重新调度这台节点上的任务重新执行。•如果有若干个不同节点上的reduce任务去获取某一台远程节点上的数据均发生了失败，那么，那台远程节点宕机的可能性就非常大。•基于信誉的探测模型：每台节点设定一个初始的信誉值（及可靠度），当其他节点的reduce任务从某个特定节点获取map结果失败后，特定节点上的信誉值将会被扣除一个惩罚值。基于信誉的探测模型时间信息空间信息当在t时刻来的fetch-error的汇报源已经在Φ(B)时，那么主要影响的是它的时间信息；当汇报源不在Φ(B)时，主要的影响因素为空间信息。实验结果•实验环境配置•负载配置实验结果在失效情况下不同检测机制的平均执行时间实验结果自适应超期时间在两种作业下的对比两个程序分别运行12次后，在每次面对一个节点宕机情况下的运行时间JavasortmonsterQuery实验结果基于信誉的探测在两种作业下的对比两个程序分别运行12次后，在每次面对一个节点宕机情况下的运行时间Javas