大数据之MapReduce应用

MapReduce应用云计算与大数据概论目标了解MapReduce计算模型的基本原理和工作流程了解Hadoop中实现并行计算的相关机制掌握MapReduce的任务调度过程掌握MapReduce的详细执行过程了解MapReduce新框架YARN的原理及运作机制MapReduce模型概述MapReduce是Google公司的核心计算模型，它将复杂的、运行于大规模集群上的并行计算过程高度地抽象到了两个函数——Map和Reduce——之中。一个MapReduce作业（Job）通常会把输入的数据集切分为若干独立的数据块，由Map任务（Task）以完全并行的方式进行处理。MapReduce框架会对Map的输出先进行排序，然后把结果输入给Reduce任务，通常作业的输入和输出都会被存储在文件系统中，而整个框架负责任务的调度和监控，并重新执行已经失败的任务。MapReduce计算模型的核心是Map和Reduce两个函数，这两个函数由用户负责实现，作用是按照一定的映射规则，将输入的键值对转换成另一个或一批键值对并输出。基于MapReduce计算模型编写分布式并行程序非常简单，程序员只需负责Map和Reduce函数的主要编码工作，而并行编程中的其它种种复杂问题，诸如分布式存储、工作调度、负载均衡、容错处理、网络通信等，均可由MapReduce框架（如Hadoop）代为处理，程序员完全不用操心。工作组件MapReduce的运行机制中主要包含以下几个独立的大类组件：Client：主要作用是提交MapReduce作业并为用户显示处理结果。ResourceManager：主要进行MapReduce作业执行的协调工作，是MapReduce运行机制中的主控节点。NameNode：文件管理系统中的中心服务器，负责管理文件系统的命名空间（元数据），维护整个文件系统的文件目录树以及这些文件的索引目录，并记录文件和目录的拥有者和权限、文件包含的数据块、块的个数以及块的副本数，以及决定数据块（Block）到具体DataNode节点的映射等。ApplicationMaster：管理在YARN内运行的应用程序实例，协调来自ResourceManager的资源。NodeManager：每一台机器框架的代理，是执行应用程序的容器，监控应用程序的资源使用情况（如CPU，内存，硬盘，网络等）并向调度器汇报。DataNode：负责处理文件系统的读写请求，在NameNode的指挥下进行数据块的创建、删除和复制。Container：YARN中资源的抽象，封装了某个节点上一定量的资源（CPU和内存两类资源）MapReduce工作流程MapReduce处理大数据集的计算过程就是将大数据集分解为成百上千的小数据集，每个（或若干个）数据集分别由集群中的一个节点（通常就是一台普通的计算机）进行处理并生成中间结果，然后这些中间结果又由大量的节点进行合并，从而形成最终结果，如图MapReduce各个执行阶段Hadoop中一个简单的MapReduce任务执行流程如下：（1）JobTracker负责在分布式环境中实现客户端任务的创建和提交；（2）InputFormat模块负责进行Map前的预处理。（3）将RecordReader处理后的结果作为Map的输入数据，然后由Map执行预先定义的Map逻辑，将处理后的键值对结果输出到临时中间文件。（4）在MapReduce流程中，为了让Reduce能并行处理Map结果，必须对Map的输出结果进行一定的排序和分割，然后再交给对应的Reduce，而这个将Map输出作进一步整理并交给Reduce的过程，就称为Shuffle。（5）Reduce执行具体的业务逻辑，即用户编写的处理数据并得到结果的业务，并且将处理结果输出给OutputFormat。Shuffle过程详解Shuffle过程是MapReduce工作流程的核心，Shuffle过程包含在Map和Reduce两端中，在Map端，Shuffle过程对Map的结果进行划分（Partition）、排序（Sort）和溢写（Spill），然后将属于同一划分的输出结果合并在一起并写到磁盘上。而在Reduce端，Shuffle又会将各个Map送来的属于同一划分的输出结果进行合并（Merge），然后对合并的结果进行排序，最后交给Reduce处理。Map端的Shuffle过程Map端的Shuffle过程简单地说，每个MapTask都有一个内存缓冲区，存储着Map的输出结果，当缓冲区快满的时候，需要将缓冲区的数据以一个临时文件的方式存放到磁盘，当整个MapTask结束后，再对磁盘中这个MapTask产生的所有临时文件做合并，生成最终的正式输出文件，然后等待ReduceTask来取数据。Reduce端的Shuffle过程ReduceTask前面的工作就是不断地拉取当前Job里每个MapTask的最终结果，然后对从不同地方拉取过来的数据不断地执行Merge（整合）操作，最终形成一个文件作为ReduceTask的输入文件，如图并行计算的实现-1数据分布存储Hadoop中的分布式文件系统HDFS由一个管理节点（NameNode）和N个数据节点（DataNode）组成，每个节点均是一台普通的计算机。在使用方面与我们熟悉的单机文件系统非常类似，同样可以新建目录、创建、复制、删除文件、以及查看文件内容等。但其在底层实现方面是把文件切割成Block（数据块），然后将这些Block分散存储在不同的DataNode上，每个Block还可以复制数份存储于不同的DataNode上，达到容错容灾的目的。NameNode则是整个HDFS的核心，它通过维护一些数据结构，记录下每一个文件被切割成了多少个Block，这些Block可以从哪些DataNode中获得以及各个DataNode的状态等重要信息。并行计算的实现-2分布式并行计算Hadoop中有的JobTracker用于调度和管理其它的TaskTracker。JobTracker可以运行在集群中任一台计算机上，但负责执行任务的TaskTracker必须运行于DataNode上，即DataNode既是数据存储节点，也是计算节点。MapReduce任务请求调度的过程包括：JobClient提交作业和JobTacker调度作业。并行计算的实现-3本地计算本地计算，即数据存储在哪一台计算机上，就由这台计算机进行这部分数据的计算，这样可以减少数据在网络上的传输，降低对网络带宽的需求。在Hadoop这样的基于集群的分布式并行系统中，计算节点可以很方便地扩充，因为它所能够提供的计算能力近乎是无限的，但是，由于数据需要在不同的计算机之间流动，故而对网络带宽的要求比较高，这种情况下，本地计算就是最有效的一种节约网络带宽的手段，业界将其形容为“移动计算比移动数据更经济”。并行计算的实现-4任务粒度把原始大数据集切分成小数据集时，需要控制好切分粒度，通常是让小数据集小于或等于HDFS中一个数据块的大小（默认是64M），这样能够保证一个小数据集位于一台计算机上，便于进行本地计算。如果有M个小数据集待处理，就启动M个Map任务，且这M个Map任务是分布于N台计算机上并行运行的。Reduce任务的数量R则可由用户指定。并行计算的实现-5PartitionPartition（划分）是可选配置，主要作用是在有多个Reduce的情况下，指定Map的结果由某一个Reduce处理，每一个Reduce都会有单独的输出文件。Partition会把Map任务输出的中间结果按键的范围划分成R份（R是预先定义的Reduce任务的个数），划分时通常使用Hash函数，这样可以保证某一范围内的键一定是由一个Reduce任务来处理，简化Reduce的过程。并行计算的实现-6Combine在Partition之前，还可以对中间结果先进行Combine（合并），即将中间结果中有相同键的键值对合并成一对。Combiner是可选择的，它的主要作用是在每一个Map执行完分析以后，在本地优先做Reduce的工作，减少中间结果中键值对的数目，从而减少在Reduce过程中的数据传输量。Combine的过程与Reduce的过程类似，但Combine是作为Map任务的一部分，在执行完Map函数后紧接着执行的。并行计算的实现-7Reduce任务从Map任务节点获取中间结果Map任务的中间结果在完成Combine和Partition之后，以文件形式保存于本地磁盘。中间结果文件的位置会被通知给主控JobTracker，由JobTracker再通知Reduce任务到哪个DataNode上去取中间结果。注意：所有Map任务产生的中间结果均按其键的范围使用同一个Hash函数划分成了R份，R个Reduce任务各自负责一段键区间。每个Reduce需要从多个Map任务节点获取落在其负责的键区间内的中间结果，然后执行Reduce函数，形成一个最终的结果文件。实例分析：WordCount如果需要统计过去10年计算机论文中出现次数最多的几个单词，以分析当前的热点研究议题是什么，那将论文样本收集完毕之后，接下来该怎么办呢？如同Java中的“HelloWorld”经典程序一样，WordCount是MapReduce的入门程序，该程序要求计算出文件中各个单词的频数，并将输出结果按照单词的字母顺序进行排序，每个单词和其频数占一行，且单词和频数之间有间隔。例如，输入一个文件，其内容如下。helloworldhellohadoophellomapreduce其符合要求的输出结果为。hadoop1hello3mapreduce1world1。上面这个应用实例，就是将文件内容切分成单词，然后将所有相同的单词聚集到一起，最后计算单词出现的次数并输出计算结果。每个拿到原始数据的机器只要将输入数据切分成单词就可以了，因此，可以在Map阶段完成单词切分的任务；另外，可将相同的单词交给一台机器来计算频数，然后输出最终结果，该任务可由Reduce阶段完成；至于将中间结果根据不同单词进行分组，然后再发送给Reduce机器的任务，则可由MapReduce中的Shuffle阶段完成。至此，上述实例的MapReduce程序就设计完成了：Map阶段完成由输入数据到单词切分的工作；Shuffle阶段完成相同单词的聚集和分发工作；Reduce阶段则完成接收所有单词并计算其频数的工作。WordCount类程序源代码-1publicclassWordCount{publicstaticvoidmain(String[]args)throwsException{Configurationconf=newConfiguration();String[]otherArgs=newGenericOptionsParser(conf,args).getRemainingArgs();if(otherArgs.length<2){System.err.println("Usage:wordcount<in>[<in>...]<out>");System.exit(2);}Jobjob=Job.getInstance(conf,"wordcount");job.setJarByClass(WordCount.class);job.setMapperClass(TokenizerMapper.class);job.setCombinerClass(IntSumReducer.class);job.setReducerClass(IntSumReducer.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);for(inti=0;i<otherArgs.length-1;i++){FileInputFormat.addInputPath(job,newPath(otherArgs[i]));}FileOutputFormat.setOutputPath(job,newPath(otherArgs[(otherArgs.length-1)]));

System.exit(job.waitForCompletion(true)?0:1);}

WordCount类程序源代码-2publicstaticclassIntSumReducerextendsReducer<Text,IntWritable,Text,IntWritable>{privateIntWritableresult=newIntWritable();

publicvoidreduce(Textkey,Iterable<IntWritable>values,Reducer<Text,IntWritable,Text,IntWritable>.Contextcontext)throwsIOException,InterruptedException{intsum=0;for(IntWritableval:values){sum+=val.get();}this.result.set(sum);context.write(key,this.result);}}

publicstaticclassTokenizerMapperextendsMapper<Object,Text,Text,IntWritable>{privatestaticfinalIntWritableone=newIntWritable(1);privateTextword=newText();

publicvoidmap(Objectkey,Textvalue,Mapper<Object,Text,Text,IntWritable>.Contextcontext)throwsIOException,InterruptedException{StringTokenizeritr=newStringTokenizer(value.toString());while(itr.hasMoreTokens()){this.word.set(itr.nextToken());context.write(this.word,one);}}}}WordCount执行过程整个过程如图（Map阶段与Reduce阶段）使用Hadoop运行程序-1执行如下命令，在HDFS中，创建input文件目录。$hadoopfs-mkdir/input在Hadoop目录下，找到文件LICENSE.txt，执行如下命令，将其放到HDFS的input目录下。$hadoopfs-putLICENSE.txt/input执行如下命令，查看是否成功上传文件。$hadoopfs-lshdfs:/inputFound1items-rw-r--r--3hadoopsupergroup154292016-09-0522:44hdfs:///input/LICENSE.txt执行如下命令，运行WordCount程序。$hadoopjarshare/hadoop/mapreduce/hadoop-mapreduce-examples-2.7.0.jarwordcount/input/output使用Hadoop运行程序-2程序运行的结果以文件的形式存放在HDFS系统的output目录，地址如下。hadoop@ubuntu:~/soft/hadoop-2.7.0$hadoopfs-gethdfs:/output将运行结果文件下载到本地，查看统计结果，如图MapReduce新框架YARN-1YARN带来了巨大的改变，它改变了Hadoop计算组件（MapReduce）切分和重新组成处理任务的方式，因为YARN把MapReduce的追踪组件切分成两个不同的部分：资源管理器和应用调度，有助于更加轻松地同时运行MapReduce或Storm这样的任务以及HBase等服务。YARN框架的原理及运作机制：重构后新的HadoopMapReduce框架被命名为MapReduceV2或者YARN，其架构如图MapReduce新框架YARN-2此次重构的基本思想是：将JobTracker的两个主要功能——资源管理和任务调度/监控——分离成单独的组件。新的ResourceManager全局管理所有应用程序计算资源的分配，而每一个应用的ApplicationMaster负责相应的调度和协调。实际上，每一个应用的ApplicationMaster都是一个详细的框架库，它结合从ResourceManager获得的资源，与NodeManager协同工作来执行并监控任务。某种意义上讲，ResourceManager就是一个纯粹的调度器，它在执行过程中不对应用进行监控和状态跟踪，同样，它也不能重启因应用失败或硬件错误而运行失败的任务。ResourceManager是基于应用程序对资源的需求进行调度的，资源包括内存、CPU、磁盘、网络等等，每一个应用程序需要不同类型的资源，因此就需要不同的容器。可以看出，这与原来的MapReduce固定类型的资源使用模型有显著区别，原来的模型会给集群的使用带来负面影响。ResourceManager提供一个调度策略插件，它负责将集群资源分配给多个队列和应用程序，调度插件可以基于现有的能力调度资源。NodeManager是每一台机器框架的代理，是执行应用程序的容器，它监控应用程序的资源使用情况（CPU、内存、硬盘、网络等）并向调度器汇报。新旧HadoopMapReduce框架比对新的YARN框架相对于旧的MapRduce框架而言，其配置文件、启停脚本及全局变量等都发生了一些变化，原框架中核心的JobTracker和TaskTracker不见了，取而代之的是ResourceManager、ApplicationMaster与NodeManager三个部分，三者的具体功能如下：ResourceManager：ResourceManager是一个中心的服务，它的工作是调度、启动每一个Job所属的ApplicationMaster，并且监控ApplicationMaster的存在情况。ResourceManager负责作业与资源的调度，它接收JobSubmitter提交的作业，按照作业的上下文（Context）信息以及从NodeManager收集来的状态信息，启动调度过程，并分配一个Container作为ApplicationMaster。ApplicationMaster：负责一个Job生命周期内的所有工作，类似旧MapReduce框架中的JobTracker。NodeManager：功能比较专一，即负责Container状态的维护，并与ResourceManager保持通信。YARN框架的优势大大减少了JobTracker（也就是现在的ResourceManager）的资源消耗，并且使监测每一个Job子任务的程序分布式部署，使程序更安全、