MapReduce基础编程mapreduce基础编程

Hadoop大数据处理实战第6章MapReduce基础与编程本章导读MapReduce是Hadoop中的大数据并行处理框架，其原理简单且易于实现。它将一个数据处理任务划分成Map和Reduce两个任务，使得用户可以在不清楚分布式计算框架内部运行机制的情况下轻松完成计算任务。本章首先介绍分布式并行计算的概念及MapReduce的含义和特点，然后介绍MapReduce的工作流程，并重点介绍Shuffle过程，最后通过单词统计和二次排序两个案例具体介绍如何利用MapReduce进行编程。学习目标020304了解分布式并行计算的概念。掌握MapReduce的基础知识。熟悉MapReduce的工作流程。掌握MapReduce的Shuffle过程。掌握MapReduce的编程方法。01MapReduce概述MapReduce的工作流程02目录CONTENTS案例解析——单词统计编程03案例解析——二次排序编程0401MapReduce概述MapReduce是Hadoop系统中最重要的计算引擎，它不仅直接支持交互式应用、基于程序的应用，还是Hive等组件的基础。MapReduce概述MapReduce概述6.1.1分布式并行计算分布式计算（distributedcomputing）可以将大任务拆分成许多小任务，然后把这些任务派发给多台计算机进行计算，最后将所有的计算结果进行汇总，以得到最终的结果。此外，分解后的任务之间互相独立，计算结果之间几乎互不影响，且实时性要求不高。1．分布式计算并行计算（parallelcomputing）又称平行计算，它是一种允许让多条指令以平行的方式同时进行计算的模式，包括时间并行和空间并行两种方式。其中，时间并行可理解为利用多条流水线同时作业，而空间并行可认为是使用多种计算资源（如多个处理器）执行并发计算，从而减少解决复杂问题所花费的时间。2．并行计算对于大多数开发人员而言，并行计算是复杂而陌生的，至于分布式计算，直接进行编程就更加困难了。而MapReduce的出现，很好地解决了分布式并行计算的难题。MapReduce概述6.1.2MapReduce的含义受Google发表的MapReduce论文的核心思想启发，HadoopMapReduce将Google的核心计算模型MapReduce进行了开源实现，使之可以在大规模集群上并行处理海量数据。通俗地讲，MapReduce是一个针对大规模集群中的分布式文件进行并行处理的计算模型。MapReduce原理简单且易于实现，其设计目标就是让不熟悉分布式并行编程的开发人员，将自己的程序轻松运行在分布式系统上。此外，MapReduce屏蔽了分布式并行计算的诸多细节，仅为用户提供MapReduce计算接口，从而使得开发人员只需进行简单的编程就可以编写并行计算程序。MapReduce概述MapReduce主要将数据处理任务拆分为Map（映射）和Reduce（规约）两个任务。MapReduce充分体现了“分而治之”的思想，即把一个复杂的任务拆分成小的任务并行处理，从而提高了任务的处理速度。某个房屋销售公司想要统计某月的销售总额，老板就会让总经理报告销售总额，而总经理则会把统计任务分配给各地区经理，让他们各自统计其管辖区域的销售总额，最后总经理将各地区的销售总额汇总后报告给老板。这就是一个典型的MapReduce任务：总经理将销售额统计任务分给各地区经理，相当于Map任务，而总经理汇总各地区的销售额，就是Reduce任务。MapReduce概述6.1.3MapReduce的特点MapReduce作为Hadoop的核心组件之一，它在数据处理方面具有自己的特色。概括而言，MapReduce在大数据处理方面具有以下几个优势：

1（1）开发简单且易于实现。MapReduce在执行分布式并行计算时，会将并行编程的繁琐细节隐藏起来，如任务调度、负载均衡、失败恢复等。所以，开发人员在设计程序时，只需要设计简单的计算逻辑，就可以完成分布式计算任务，这就大大降低了分布式程序的编写难度。MapReduce概述

2（2）良好的可扩展性。MapReduce具有良好的可扩展性，这意味着当集群计算资源不足时，可以通过动态增加节点的方式实现弹性计算。

3如果集群中的某计算节点出现故障，使得作业执行失败，MapReduce可自动将作业分配到可用的计算节点上重新执行。（3）高容错性。MapReduce概述MapReduce也存在以下局限性：（1）计算局限。MapReduce适用于海量数据的离线批处理，但不能像Oracle数据库那样在毫秒时间内返回计算结果。因此，它不适合数据事务处理或单一请求处理。如果用户需要大数据的毫秒级响应，可以考虑HBase应用程序。此外，MapReduce中的输入数据是静态的，所以它也不适合流式数据计算。如果用户想要进行流式数据计算，可以使用Storm、SparkStreaming和Flink等流式计算框架，并且它们同MapReduce一样，都可以在YARN上运行。（2）性能局限。Map任务与Reduce任务存在严格的依赖关系。Map任务的中间结果存储在本地磁盘上，然后Reduce任务需要从磁盘上获取Map计算的中间结果并将其作为输入。这样一来，就会产生大量的磁盘I/O，从而使得计算性能降低。不过，Spark与MapReduce不同，它在内存中运算，这样就避免了磁盘I/O。（3）应用局限。MapReduce不适合一般的Web应用，因为这些应用只是简单的数据访问且每次访问请求所需要的资源非常少，同时还要满足高并发访问需求。02MapReduce的工作流程对于某些简单的数据处理任务，如文件中的数据只做简单的数据格式转换或切分等，可能只需要执行Map任务就够了，然后将Map任务输出的结果直接存储到HDFS中。但对于大多数复杂的计算任务来说，往往离不开Reduce任务。用户需根据实际的业务需求设置合适的Reduce任务数（默认为1），来完成数据处理任务。MapReduce的工作流程可将MapReduce的工作流程分为5个阶段输入分片和数据格式化Map过程Shuffle过程Reduce过程结果输出MapReduce的工作流程在MapReduce计算过程中，以<key,value>键值对的形式进行数据传递或计算，MapReduce的简易数据处理流程如下图所示。MapReduce的简易数据处理流程MapReduce的工作流程MapReduce的工作流程6.2.1输入分片和数据格式化在执行Map任务之前，MapReduce会将存储在分布式文件系统中的大规模数据集切分成独立的输入分片（InputSplit），并且每一个输入分片对应着一个Map任务。也就是说，有多少个输入分片就会存在多少个Map任务。输入分片是一个逻辑概念，它对输入数据集的切分不是物理意义上的切分，而是对数据的逻辑结构进行切分。每个输入分片存储的并不是真实数据，而是指向分片数据的引用。例如，输入分片中存储了一些元数据信息，包括起始位置、数据长度、数据所在节点等。此外，由于Map任务的输入数据要求是键值对的形式，所以需要对输入分片进行格式化，即将输入分片处理成<key1,value1>形式的数据，然后再传递给Map任务。MapReduce的工作流程6.2.2Map过程Map过程利用map()函数来处理数据，map()函数接收<key1,value1>形式的数据输入。经过map()函数处理后，获得一系列<key2,value2>形式的输出。其中，map()函数中具体的数据处理方法可以由用户自己定义。

map()函数的调用MapReduce的工作流程6.2.3Shuffle过程通常，map()函数的输出并不会直接交给Reduce任务，而是需要经过一系列处理，如分区（Partition）、排序（Sort）、合并（Combine）、溢写（Spill，将内存中的数据写入磁盘）、归并（Merge）等，然后将处理后的数据作为Reduce任务的输入。这一系列处理过程称为Shuffle（洗牌）过程。

MapReduce的Shuffle过程MapReduce的工作流程（1）map()函数的输出并不会立即写入磁盘，MapReduce会为每个Map任务分配一个环形内存缓冲区（bufferinmemory），用于存储map()函数的输出。（2）在将环形内存缓冲区中的数据写入磁盘之前，需要对数据进行分区、排序和合并（可选）操作。①分区操作的主要目的是将数据均匀地分配给Reduce任务，以实现MapReduce的负载均衡，从而避免单个Reduce任务的压力过大。②排序操作是MapReduce的默认操作，主要是将Map任务的输出按key进行排序。排序操作在分区操作之后执行，因此，每个分区中的数据都是有序的。③排序结束后，用户可根据实际需求选择是否要执行合并操作。不过，只有预先定义了Combine()函数，才会执行合并操作，从而减少溢写的数据量。所谓合并操作，就是将具有相同key的<key,value>的value加起来。例如，具有相同键的<'a',1>和<'a',1>两个键值对，经过合并操作之后，得到的结果为<'a',2>，这样一来，键值对的数量就减少了。1．Map端的Shuffle过程MapReduce的工作流程（3）环形内存缓冲区中的数据一旦达到阈值，一个后台线程便开始把数据溢写到本地磁盘的临时文件（即溢写文件）中。在溢写到磁盘的过程中，Map的输出仍然不断地写到环形内存缓冲区中。不过，当整个环形内存缓冲区被数据占满时，Map任务就会被阻塞，直到写磁盘过程完成，才可以向环形内存缓冲区继续写数据。（4）由于此时Map任务并未结束，系统需要将所有溢写文件中的数据进行归并（从磁盘到磁盘以分区排序来归并数据），以生成一个大的溢写文件（数据已分区且有序）。归并操作就是将相同key的value归并成一个集合，形成新的键值对。例如，具有相同键的键值对<'a',1><'a',2><'a',5>，经过归并操作之后，得到的键值对为<'a',{1,2,5}>。文件归并操作完成后生成最终的Map任务输出文件，文件保存在Map任务所在节点的本地磁盘上，Map任务执行结束。MapReduce的工作流程（1）在一个MapReduce作业中，通常会启动多个Map任务，并且由于每个Map任务处理的数据量不同，任务结束时间也不同。不过，一旦有Map任务结束，与其相关的Reduce任务就会去复制输出文件。系统会根据Reduce任务数来启动相同数量的复制线程（Fetcher），这些复制线程能够并行复制Map任务的输出文件。（2）Reduce任务将复制获得的文件存放在自身所在节点的缓存中，当缓存中的数据达到阈值，即需要溢写到磁盘时，Reduce任务会将复制数据进行归并排序（MergeSort），生成溢写文件。如果生成了多个溢写文件，则需要多次执行归并操作，再将数据输入到reduce()函数。2．Reduce端的Shuffle过程MapReduce的工作流程6.2.4Reduce过程Reduce任务的输出结果可以经过输出格式化后再输出到文件系统中，并且每个作业输出结果文件默认以“part-r-00000”开始，并用后5位数递增的方式命名。Reduce任务接收归并排序后的数据流，并对已有序的相同key的键值对调用一次reduce()函数。其中，Reduce任务的输入是<key2,list(value2)>形式的中间结果，输出的是<key3,value3>形式的计算结果。03案例解析——单词统计编程案例解析——单词统计编程基于MapReduce的编程之所以难度不大，是因为MapReduce已经为开发人员解决了诸如分布式存储、负载均衡、容错管理、网络通信、工作调度等难题。开发人员只需分别实现Map任务和Reduce任务中的map()函数和reduce()函数即可，而这两个函数来自MapReduceAPI提供的Mapper类和Reducer类。MapReduce主要处理HDFS中的文件，而HDFS中的文件格式多种多样，且文件的大小也不固定，大的文件可达几百GB，而小的文件却只有几个KB。因此，处理这样的输入数据时需要对其进行格式化。此外，为了方便使用输出结果，同样也需要对输出数据进行格式化。除了输入、输出数据的格式化，还可以根据实际问题优化数据处理过程，如数据分区、数据压缩、数据合并（Map端的合并操作）及数据过滤等。案例解析——单词统计编程6.3.1案例描述假设在hdfs://hadoop0:9000/jqe/wc/data目录中存在两个文本文件（文件中仅有字符串和空格），分别是words1.txt和words2.txt，现需计算出这两个文件的单词频数。要求输出结果按单词的字母顺序进行排序，并输出到两个文件中（以字母“g”进行拆分）。另外，还需要对输出数据进行格式化，要求单词与词频之间以“~~~~~~”隔开。其中，words1.txt的内容如下：thisisagooddogthatisagoodcatwords2.txt的内容如下：thisisabaddogthatisabadcat案例解析——单词统计编程输出结果应分为两个文件（系统自动命名），分别是part-r-00000和part-r-00001。其中，part-r-00000的内容如下：a~~~~~~4bad~~~~~~2cat~~~~~~2dog~~~~~~2part-r-00001的内容如下：good~~~~~~2is~~~~~~4that~~~~~~2this~~~~~~2案例解析——单词统计编程6.3.2设计思路在单词统计任务中，可以将大的数据集切分成小的数据集，且各数据集之间相互独立，方便并行处理。此外，各个单词之间的频数不具有相关性，可以将不同的单词分发到不同的机器上处理。由此可以看出，单词统计任务的解决思路完全贴合MapReduce的编程思想。根据分析，单词统计程序可分为7个模块，分别是输入、输入分片及其格式化、设计map()函数、分区、设计reduce()函数、输出格式化、输出。案例解析——单词统计编程指定输入文件位置。将两个文件切分成两个输入分片，然后对输入分片进行格式化，如文件按行分解，从而形成以key为偏移量、value为行内容的键值对。设计map()函数，将键值对的value按空格分解成一个个单词（word），生成<word,1>形式的键值对。设计分区函数，将key的首字母小于“g”的键值对划分为一个分区，其余的划分为另外一个分区。设计reduce()函数，将输入的<word,list(value)>形式（如<"is",{1,1}>）的数据进行汇总，生成以key为单词、value为单词频数的键值对。系统默认的输出格式为“单词+空格+单词频数”的形式，若要输出特定样式的数据，需要对输出结果进行格式化，使得单词与单词频数之间以“~~~~~~”间隔。指定单词统计程序的输出结果文件位置。案例解析——单词统计编程6.3.3执行过程在给出的案例中，输入文件为两个很小的文本文件，单个文件的数据没有达到需要切分的程度，所以可将每个文件作为独立的分片。此外，还需要对输入分片进行格式化操作，形成<key1,value1>形式的数据流。1．输入分片及其格式化单词统计的输入分片及其格式化key1为偏移量，从0开始，每读取一个字符（包括空格）就增加1；value1为每行文本内容（字符串形式）。案例解析——单词统计编程map()函数将接收到的<key1,value1>形式的输入数据流，按空格进行拆分，输出结果为<key2,value2>形式的数据。2．Map过程key2为字符串形式的单词；value2的值为1，表示单词数为1。单词统计的Map过程案例解析——单词统计编程由于Reduce要求输入数据有序，所以map()函数的计算结果需要经过处理（如分区、排序、归并），才可以作为reduce()函数的输入。于是，将多个Map任务的<key2,value2>形式的输出，处理成<key2,list(value2)>形式的中间结果3．Shuffle过程单词统计的Shuffle过程案例解析——单词统计编程reduce()函数接收<key2,list(value2)>形式的数据流，对相同单词的值集合进行计算，汇总出单词出现的总次数。4．Reduce过程单词统计的Reduce过程案例解析——单词统计编程用户可以根据需求，更改单词统计结果的输出格式。5．输出单词统计的输出结果格式化案例解析——单词统计编程6.3.4编程实现在Eclipse开发工具中新建一个Map/Reduce项目“WordCount”（即单词统计项目），然后根据单词统计的设计思路进行MapReduce编程。使用MapReduceAPI中的org.apache.hadoop.mapreduce.lib.input.TextInputFormat进行输入格式化；自定义WordCountMapper类实现单词统计的map()函数；自定义WordCountPartitioner类实现对单词统计结果的分区存储；自定义WordCountReducer类实现单词统计的reduce()函数；自定义WCFileOutputFormat类实现对数据结果的格式化。编写WordCountDriver类（驱动类）去调用其他类定义的方法。案例解析——单词统计编程自定义的WordCountMapper类继承了org.apache.hadoop.mapreduce.Mapper类。WordCountMapper类是map()函数的执行者，用户需要根据数据处理需求重写map()方法。1．自定义WordCountMapper类创建WordCountMapper类定义了值为1的“one”常量，表示每个单词都是统计一次，然后又定义了一个“words”字符串数组，用于存放每行分解的单词，最后将“one”和“words”作为参数传给write()方法，相当于map()函数的一次输出。WordCountMapper类的代码参照课本。案例解析——单词统计编程自定义的WordCountPartitioner类继承了org.apache.hadoop.mapreduce.Partitioner类，然后重写getPartition()方法。该方法中使用了正则表达式，并按key对map()函数输出的键值对进行划分。例如，将key的字母小于“g”的键值对发送给一个Reduce任务，其他的键值对发送到另一个Reduce任务。WordCountPartitioner类的代码参照课本。2．自定义WordCountPartitioner类自定义的WordCountReducer类继承了org.apache.hadoop.mapreduce.Reducer类，并重写了reduce()方法。相同单词的键值对会在同一个Reduce任务中处理，该方法中定义了一个“count”整型变量，循环遍历一个键值对的valuelist，用于保存value累加的结果。WordCountReducer类的代码参照课本。3．自定义WordCountReducer类案例解析——单词统计编程自定义的WCFileOutputFormat类继承了org.apache.hadoop.mapreduce.lib.output.FileOutputFormat类，并重写了getRecordWriter()方法。该方法中定义了一个文件输出流对象，用于将结果输出到HDFS文件中。WCFileOutputFormat类的代码参照课本。4．自定义WCFileOutputFormat类自定义的WordCountDriver类为Job指定了输入文件位置、输入格式化类、自定义Partitioner类、自定义Mapper类、自定义Reducer类、自定义OutputFormat类及输出文件位置，其代码参照课本。5．自定义WordCountDriver类案例解析——单词统计编程6.3.5运行程序运行WordCount程序有两种方法：利用EclipseHadoop插件在本地提交任务到集群上运行；将编写完成的程序打包成jar包并上传到集群上运行。1．利用EclipseHadoop插件1编写一个perties文件（用于记录MapReduce的运行日志），并将其放入WordCount项目的“src”文件夹，然后在Eclipse中手动刷新一下。案例解析——单词统计编程2右击WordCountDriver类，从弹出的快捷菜单中选择“RunAs”→“RunonHadoop”选项，以该方式编译运行程序。程序运行完成后，会在“hdfs://hadoop0:9000/jqe/wc/out/”路径下生成相关文件：_SUCCESS、part-r-00000和part-r-00001，单词统计结果如下图所示。本地提交任务到集群上运行的结果案例解析——单词统计编程2．将项目打包成jar包1将WordCount项目打包成一个jar包：在“ProjectExplorer”窗口中右击“WordCount”项目，从弹出的快捷菜单中选择“Export...”选项，打开“Export”对话框，选择“Java”→“JARfile”选项，然后单击“Next”按钮，弹出“JARExport”对话框，指定jar包的保存位置和名称（如E:\WCMR.jar），然后单击“Finish”按钮。2通过WinSCP软件将jar包“WCMR.jar”上传到集群节点（如hadoop0）的本地目录（如“/opt”）。案例解析——单词统计编程3在hadoop0节点中执行以下命令，运行WordCount程序：#hadoopjar/opt/WCMR.jardriver.WordCountDriverWordCount程序在集群上的运行过程如下图所示。单词统计结果与第1种方法相同。WordCount程序在集群上的运行过程案例解析——单词统计编程6.3.6数据优化数据压缩对于海量数据处理具有重要的意义。在MapReduce作业中，存在数据压缩的情况有3种：1．数据压缩①输入数据可以是压缩文件；②Map输出的结果可以是压缩数据；③Reduce输出的最终结果可以是压缩文件。MapReduce程序的输入、输出是压缩文件易于理解。另外，若用户选择在Map阶段对数据进行压缩，就能够减少在Reduce阶段复制Map数据结果的数据量。但值得注意的是，数据压缩不适合小文件多的MapReduce任务。案例解析——单词统计编程设置Map输出为压缩数据有两种方法。在配置文件mapred-site.xml中设置压缩参数，具体可通过配置press属性（是否开启压缩）和press.codec属性（指定压缩格式）来进行。在Job中声明，如在WordCountDriver类中添加如下代码：conf.setBoolean("press",true);conf.setClass("press.codec",BZip2Codec.class,CompressionCodec.class);案例解析——单词统计编程2．Map端的合并操作Map端的合并操作没有默认的实现，需要在Job中显式设置才能起作用。如果设置了Map端的合并操作，那么每个运行的Map任务中都会执行合并操作，且仅仅只处理自身节点生成的数据。Map端的合并操作相当于对输入Reduce任务的数据进行预处理，即将相同key的键值对合并。如此一来，Map任务的输出数据更紧凑且数据量变小，从而减少了Reduce任务复制Map任务计算结果的网络带宽和Reduce上的负载。若希望在单词统计程序中设置Map端的合并操作，可在WordCountDriver类中增加以下代码：job.setCombinerClass(WordCountReducer.class);案例解析——单词统计编程下图显示了在MapReduce的Shuffle过程中设置了Map端的合并操作后的数据流的形式，显然，它与未设置合并操作的数据流形式有所不同。

配置Map端合并操作的Shuffle过程根据实际数据处理任务的不同，数据过滤操作使用户可以只针对有价值的数据进行计算，从而大大减少数据的计算量。尤其是针对海量数据进行处理时，此项功能非常有用。3．数据过滤案例解析——单词统计编程在统计单词的MapReduce程序中，可以通过对Map任务的输出结果进行过滤来减少进入Reduce任务的计算量，如不统计“am”“is”“are”这样的词。为此，可更改WordCountMapper类的map()方法，代码如下：publicvoidmap(LongWritablek1,Textv1,Mapper<LongWritable,Text,Text,IntWritable>.Contextcontext)throwsIOException,InterruptedException{ Stringline=v1.toString(); //获得行

String[]words=line.split(""); //按空格切分行，获得单词数组

List<String>passwords=newArrayList<String>(); passwords.add("am"); passwords.add("is"); passwords.add("are"); for(Stringw:words){ //输出键值对<单词,1> word.set(w); if(!passwords.contains(word)){ context.write(word,one); } }}04案例解析——二次排序编程案例解析——二次排序编程MapReduce在传递<key,value>键值对时，默认按照key进行排序，但有时候除了key以外，还需要根据value或value中的某一个字段进行排序，基于这种需求进行的自定义排序称为“二次排序”。案例解析——二次排序编程6.4.1案例描述假设文本文件user.txt中存放着某公司的职工信息，每行均包含两个字符串，并以空格分隔。其中，第1个字符串表示部门ID，第2个字符串表示职工姓名。现要求先按部门升序排列，然后相同部门的职工姓名以字典序排列，并且部门之间以“------------------”分隔开。6.4.2设计思路Map任务的输出按key进行排序，如果将本例中的第1个属性作为Map任务输出键值对的key，第2个属性作为value，那么二次排序任务还要对value排序，但MapReduce没有提供该方法。因此，可以将key和value结合起来，形成newkey组合键，设置value为NullWritable类型的占位符，则Map任务的输出结果为<newkey,null>，即<<key,value>,null>。案例解析——二次排序编程二次排序的过程分为输入、输入分片及其格式化、Map过程、排序过程、分组过程、Reduce过程和输出过程。（1）二次排序程序的输入文件只有一个，因此可将user.txt文件作为一个输入分片，然后进行格式化。二次排序的输入分片及其格式化案例解析——二次排序编程（2）map()函数将接收到的输入数据流按行内容以空格进行拆分，输出结果为<<key,value>,null>形式的数据。其中，key为文本中第1个属性，value为文本中第2个属性。二次排序的Map过程（3）将map()函数的输出键值对按key进行排序，当key相同时，按value进行排序。二次排序的map()函数结果排序案例解析——二次排序编程（4）在Reduce端设置分组方法，按key进行分组。二次排序的分组操作（5）red