大数据技术原理与操作应用 第4章 MapReduce 分布式计算系统

第4章MapReduce分布式计算系统1MapReduce运行模式MapReduce运行流程MapReduce编程案例分析使用MapReduce实现反向索引MapReduce基本思想MapReduce处理流程MapReduce编程模型2✎学习目标理解掌握掌握掌握理解MapReduce的核心思想12掌握MapReduce的编程模型4掌握MapReduce常见编程组件的使用3掌握MapReduce的工作原理本章将对MapReduce进行系统的介绍,让读者对MapReduce有一个全面的认识。章节概要HadoopMapReduce是一个面向大规模数据集的简单易用的并行计算模型,它极大地方便了编程人员在不会分布式并行编程的情况下,将自己的程序运行在分布式系统上。

4.1 MapReduce介绍4.1.1 MapReduce基本思想MapReduce是Google提出的大规模并行计算框架,应用于大规模廉价集群上的大数据并行处理。MapReduce采用“分而治之”的设计思想,将输入的大量数据(这些数据之间不存在或有较少的依赖关系)采用一定的划分方法进行分片,然后将一个数据分片交由一个任务去处理,这些任务并行计算,最后再汇总所有任务的处理结果。MapReduce将大数据计算任务划分成多个子任务,然后由各个分节点并行计算,最后通过整合各个节点的中间结果,将各个子任务的结果进行合并,得到最终结果。MapReduce借助函数式编程设计思想,将大数据处理过程主要拆分为Map(映射)和Reduce(归约)两个模块。Map(映射)用来将输入的大量键值对映射成新的键值对,Reduce(归约)负责收集整理Map操作生成的中间结果,并进行输出。

4.1 MapReduce介绍4.1.1 MapReduce基本思想MapReduce是一个并行计算与运行软件框架,能自动完成计算任务的并行化处理,自动划分计算数据和计算任务,在集群节点上自动分配和执行任务以及收集计算结果,为程序员隐藏系统底层细节。这样程序员就不需要考虑数据的存储、划分、分发、结果收集和错误恢复等诸多细节问题,这些问题都交由系统自行处理,大大减少了软件开发人员的负担。

4.1 MapReduce介绍4.1.2 MapReduce处理流程MapReduce是一种并行编程模型,将计算分为两个阶段:Map阶段和Reduce阶段。首先将输入数据划分成多个块,由多个Map任务并行计算。MapReduce对Map任务的结果进行聚集和混洗,然后提供给Reduce任务作为其输入数据集。最终通过合并Reduce任务的输出得到最终结果。MapReduce数据处理流程如图4.1所示。图4.1MapReduce数据处理流程

4.1 MapReduce介绍4.1.1 MapReduce基本思想①从存储系统中读取输入文件内容,存储系统可以是本地文件系统或者HDFS文件系统等。对输入文件的每一行解析成一个<key,value>对,在默认情况下,key表示行偏移量,value表示这行的内容。②每一个<key,value>对调用一次map函数。程序员需要根据实际的业务需要重写map()方法,对输入的<key,value>对进行处理,转换为新的<key,value>对输出。(1)Map任务处理

4.1 MapReduce介绍4.1.1 MapReduce基本思想①对Map输出的<key,value>对进行分区,并将结果通过网络复制到不同的Reducer节点上。②将不同分区的数据按照key进行排序,相同key的value放到一个集合中,形成新的键值对,即<key,list(value)>对,记为<key,VALUE>。(2)Shuffle与Sort

4.1 MapReduce介绍4.1.1 MapReduce基本思想①调用Reduce函数处理前面得到的每一个<key,VALUE>。程序员需要根据实际的业务需要重写reduce()方法。②将Reduce函数的输出保存到文件系统中。(3)Reduce任务处理

4. 2MapReduce运行机制4.2.1MapReduce编程模型简介一个MapReduce作业通常将输入的数据集拆分成多个独立的块,这些块被Map任务以并行的方式进行计算。MapReduce将Map任务的输出进行排序,然后将排序后的结果作为Reduce任务的输入。作业的输入和输出数据均被存储在文件系统中。MapReduce框架负责任务调度、监控和重新执行失败的任务。HadoopMapReduce为用户提供了五个可编程组件,分别是InputFormat、Mapper、Partitioner、Reducer和OutputFormat。还有一个组件称为Combiner,实际上是一个局部的Reducer,通常用于MapReduce程序性能方面,不属于必备组件。一般情况下,用户只需要编写Mapper和Reducer类即可,其他类已由MapReduce实现,可以直接使用。

4. 2MapReduce运行机制4.2.2 MapReduce运行模式Hadoop的MapReduceJob的运动模式可以分为本地运行模式和集群运行模式。本地运行模式在编写MapReduce程序时不带集群的配置文件,也就是MapReduce程序不要配置YARN作为运行框架,配置“=local”,就可以实现程序的本地运行。MapReduce的本地运行模式无须启动远程的Hadoop集群,MapReduce程序会被提交给本地执行器LocalJobRunner在本地以单进程的形式运行。输入数据及输出结果可以放在本地文件系统,也可以放在HDFS上。本地运行模式非常便于进行业务逻辑调试,只要在Eclipse中设置断点即可。

4. 2MapReduce运行机制4.2.2 MapReduce运行模式(2)集群运行模式首先需要启动YARN,Job

会提交到YARN框架中去执行,访问“http://master:8088”可以查看Job执行状态。在MapReduce的集群运行模式下,MR程序将会提交给YARN集群Resourcemanager,分发到多个节点上并发执行。输入数据和输出结果一般位于HDFS文件系统。将MapReduce程序提交集群的实现方法有以下几种:①将程序打包成jar包,上传到服务器,然后在集群的任意节点上调用Hadoop命令启动集群执行。②在Linux的Eclipse中直接运行main方法,将程序提交到集群中去运行,但采用此种方法,项目中要带YARN的配置。③在Windows的Eclipse中直接运行main方法,也可以提交给集群去运行,但需要作更多的修改。

4. 2MapReduce运行机制4.2.3MapReduce运行流程

4. 2MapReduce运行机制4.2.3MapReduce运行流程(1)MapTask工作原理MapTask负责Map阶段的整个数据处理流程,Map阶段并行度由客户端提交作业时的切片个数决定。①Split阶段。MapReduce从文件系统中读取文件后,会首先对读取的文件进行输入分片(inputsplit)的划分。输入分片存储的并非数据本身,而是一个分片长度和一个记录数据的位置的数组。一般情况下,以HDFS的一个块的大小作为一个分片(也可以按split设置值来切片)。②一个输入分片分配给一个Map任务。Map任务分配完成后,再对传进来的分片进一步分解成一批键值对(<key,value>对),每一个键值对调用程序员事先编写的Map函数进行逻辑处理,输出新的<key,value>对。③数据处理完成后,会调用collect()函数进行结果的收集和输出。collect()函数将新生成的<key,value>对进行分区,并写入一个环形内存缓冲区中(默认为100MB)。

4. 2MapReduce运行机制4.2.3MapReduce运行流程(1)MapTask工作原理④溢写。当环形内存缓冲区满后(一般为写满缓冲区大小的80%),MapReduce会将数据写到本地磁盘上生成一个临时文件。首先,对需要溢写的缓存区内的数据进行分区,分区的个数由Reduce任务的个数决定,有多少个Reduce任务就划分为多少个分区,这样可以保证Reduce任务分到均衡的数据。然后对每个分区中的数据进行排序,经过排序后,数据按分区聚集在一起,且同一分区内所有数据按照key有序。最后进行溢写操作。如果用户设置了Combiner,则还需要将排序后的结果以分区为单位合并成大文件,以避免同时打开大量文件和同时读取大量小文件产生的随机读取带来的开销。

4. 2MapReduce运行机制4.2.3MapReduce运行流程(2)Shuffle工作原理Shuffle是MapReduce最为关键的一个阶段,主要负责将Map端生成的数据按键排好序传递到Reduce端,包括两个部分,即Map端的Shuffle和Reduce端的Shuffle。1)Map端的Shuffle将Map输出的<key,value>对放到环形缓冲区中,当缓冲区空间写满80%时,准备将缓冲区中数据写入到磁盘,这样可以保证写入到内存缓冲区和写入到磁盘并行进行,而不用中止Map。溢写之前,先按照分区和key等对数据进行排序,然后按分区将数据写入临时文件。如果有必要,还会进行合并操作,也就是将按分区输出的小文件合并成大文件,以提高效率。

4. 2MapReduce运行机制4.2.3MapReduce运行流程(2)Shuffle工作原理2)Reduce端的ShuffleReduce任务通过HTTP向各个Map任务复制它所需要的数据,Map任务会在内存或磁盘上对数据进行合并,Reduce任务复制过来的数据有些放在内存中,而有些放在磁盘上。Map输出数据已经是有序的,Reduce会对Map传递过来的数据不断地进行合并。一般Reduce端是一边复制数据一边合并数据,复制和合并是重叠的。合并完成后,便得到了Reduce端的输入文件。

4. 2MapReduce运行机制4.2.3MapReduce运行流程(3)ReduceTask工作原理①ReduceTask负责Reduce阶段的整个数据处理流程。ReduceTask的并行度可以通过“job.setNumReduceTasks()”进行手动设置。ReduceTask默认值是“1”,输出文件个数为1个。若ReduceTask设置为“0”,表示没有Reduce阶段,输出文件个数与Map个数一致。②ReduceTask从各个MapTask上通过网络远程复制数据,每个Map传来的数据都是有序的。针对复制过来的数据,如果数据量较小,则直接存放在内存中;如果数据量大小超过一定阈值,则写到磁盘上。③为了避免内存使用过多或磁盘上存在大量小文件,在ReduceTask进行拷贝的同时,会启动后台线程对内存和磁盘上的文件进行合并。因此,Reduce端一边复制数据,一边合并数据。

4. 2MapReduce运行机制4.2.3MapReduce运行流程(3)ReduceTask工作原理④为了将key相同的数据聚集在一起,ReduceTask会对所有数据进行一次归并排序,将相同key的value值放到一个列表中,生成形如<key,list(value)>的键值对。⑤对每一个键值对调用程序员事先编写的Reduce()函数进行计算,并将结果输出到文件系统中。

4. 2MapReduce运行机制4.2.4 MapReduce编程案例WordCount分析单词统计(WordCount)是MapReduce经典案例之一,用于统计大量文件中每个单词出现的次数,本节通过分析一个单词统计案例来深入剖析MapReduce编程模型。WordCount的目标是将多个inputsplit文件输入最终得到文件的词频统计,如图4.3所示。图4.3统计大量文件中的单词词频

4. 2MapReduce运行机制4.2.4 MapReduce编程案例WordCount分析1)文件转换图4.4文件转换成<key,value>对过程(1)Map阶段

4. 2MapReduce运行机制4.2.4 MapReduce编程案例WordCount分析2)map()方法作用图4.5 map()方法的转换过程(1)Map阶段

4. 2MapReduce运行机制4.2.4 MapReduce编程案例WordCount分析2)map()方法作用(1)Map阶段public

static

classWordCountMap

extendsMapper<LongWritable,Text,Text,IntWritable>{

private

final

IntWritable

one=new

IntWritable(1);

privateTextword=newText();

@Override

protected

voidmap(LongWritable

key,Textvalue, Mapper<LongWritable,Text,Text,IntWritable>.Contextcontext)

throws

IOException,InterruptedException{ Stringline=value.toString();

StringTokenizer

token=new

StringTokenizer(line);

while(token.hasMoreTokens()){

word.set(token.nextToken());

context.write(word,one); } } }

4. 2MapReduce运行机制4.2.4 MapReduce编程案例WordCount分析3)Map端的Shuffle(1)Map阶段图4.6Map端的Shuffle过程

4. 2MapReduce运行机制4.2.4 MapReduce编程案例WordCount分析1)Reduce端的Shuffle(2)Reduce阶段图4.7Reduce端的Shuffle过程

4. 2MapReduce运行机制4.2.4 MapReduce编程案例WordCount分析2)reduce()方法作用(2)Reduce阶段图4.8Reduce()方法的作用过程

4. 2MapReduce运行机制4.2.4 MapReduce编程案例WordCount分析2)reduce()方法作用(2)Reduce阶段public

static

class

WordCountReduce

extendsReducer<Text,IntWritable,Text,IntWritable>{

@Override

protected

voidreduce(Textkey,Iterable<IntWritable>values, Reducer<Text,IntWritable,Text,IntWritable>.Contextcontext)

throws

IOException,InterruptedException{

int

sum=0;

for(IntWritable

val:values){

sum+=val.get(); }

context.write(key,new

IntWritable(sum)); }}

4. 2MapReduce运行机制4.2.4 MapReduce编程案例WordCount分析(3)作业设置

public

static

voidmain(String[]args)

throws

IOException,ClassNotFoundException,InterruptedException{

//Configuration类代表作业的配置 Configurationconf=newConfiguration(); Jobjob=newJob(conf);

//指定主类为WordCount.class

job.setJarByClass(WordCount.class);

job.setJobName("wordcount");

//设置输出key类型为Text，输出值类型为IntWritable

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

//指定Mapper类和Reducer类

job.setMapperClass(WordCountMap.class);

job.setReducerClass(WordCountReduce.class);

//设置输入输出格式类

job.setInputFormatClass(TextInputFormat.class);

job.setOutputFormatClass(TextOutputFormat.class);

//设置输入数据文件路径和输出文件存放路径

FileInputFormat.addInputPath(job,newPath("hdfs://master:9000/wordcounttest/word.txt"));

FileOutputFormat.setOutputPath(job,newPath("hdfs://master:9000/wordcounttest/output"));

//提交作业并等待执行完成

job.waitForCompletion(true); }

4. 3案例:使用MapReduce实现反向索引4.3.1Windows下Eclipse远程开发环境搭建(1)安装配置JDK1）下载JDK1.8，注意版本正确。2)环境变量设置①配置JDK环境变量在“计算机”上单击鼠标右键,选择“属性”,在弹出的对话框中单击“高级系统设置”,如图4.9所示。图4.9 高级系统设置

4. 3案例:使用MapReduce实现反向索引4.3.1Windows下Eclipse远程开发环境搭建(1)安装配置JDK在“系统属性”对话框的“高级”选项卡中,单击“环境变量”,如图4.10所示。图4.10系统属性

4. 3案例:使用MapReduce实现反向索引4.3.1Windows下Eclipse远程开发环境搭建(1)安装配置JDK在“环境变量”对话框中,新建系统变量“JAVA_HOME”,输入值为JDK的安装路径,如图4.11所示。图4.11新建系统变量“JAVA_HOME”

4. 3案例:使用MapReduce实现反向索引4.3.1Windows下Eclipse远程开发环境搭建(1)安装配置JDK编辑系统变量CLASSPATH,如果没有该变量,则单击“新建”按钮进行创建。在编辑框中,输入变量值为JDK的安装目录的lib目录和jre下的lib目录。注意,第一个英文句号表示当前目录,不能省略。如图4.12所示。图4.12编辑系统变量CLASSPATH

4. 3案例:使用MapReduce实现反向索引4.3.1Windows下Eclipse远程开发环境搭建(1)安装配置JDK编辑系统变量Path,在变量值的前面添加JDK安装目录的bin目录。注意,末尾的分号不能省略。如图4.13所示。至此,JDK安装完成。图4.13编辑系统变量Path

4. 3案例:使用MapReduce实现反向索引4.3.1Windows下Eclipse远程开发环境搭建(1)安装配置JDK②测试JDK是否安装成功JDK安装完成后需要测试其是否安装成功。按“Win+R”键弹出“运行”对话框,输入“cmd”,在命令提示符中输入“java-version”查看JDK版本是否正确,如图4.14所示。图4.14查看JDK版本

4. 3案例:使用MapReduce实现反向索引4.3.1Windows下Eclipse远程开发环境搭建(1)安装配置JDK②测试JDK是否安装成功输入“javac”和“java”,如果能正确输出信息,则JDK配置成功,如图4.15和4.16所示。如果输出“Commandnotfound”,说明JDK配置失败,需要重新配置。JDK的配置非常重要,一定要保证Java环境变量配置正确。图4.16执行java命令图4.15执行javac命令

4. 3案例:使用MapReduce实现反向索引4.3.1Windows下Eclipse远程开发环境搭建(2)配置Eclipse在Windows上远程开发MapReduce项目需要使用插件“hadoop-eclipse-plugin-2.7.7.jar”(2.7.7表示对应Hadoop版本号,需保持一致)。将下载好的插件放入Eclipse安装目录下的dropins文件夹中。以管理员身份运行解压缩软件将“hadoop-2.7.7.tar.gz”解压到Windows下,由于Linux文件格式和Windows不一样,在解压过程中可能会有错误提示,可以忽略。将bin目录下的hadoop.dll和winutils.exe(可以到网上自行下载)这两个文件复制到“C:\Windows\System32\”目录下。

4. 3案例:使用MapReduce实现反向索引4.3.1Windows下Eclipse远程开发环境搭建(2)配置Eclipse重新打开Eclipse,单击Windows→Preferences,在左侧窗口中单击“HadoopMap/Reduce”选项,在窗口右侧设置Hadoop安装路径,如图4.17所示。注意,Hadoop的安装路径既不能有空格,也不能有中文。图4.17配置Windows下Hadoop安装路径

4. 3案例:使用MapReduce实现反向索引4.3.1Windows下Eclipse远程开发环境搭建(2)配置Eclipse配置Map/ReduceLocations。单击Eclipse中Window→Perspective→OpenPerspective→Other,单击“Map/Reduce”,在下方窗口中单击“Map/ReduceLocations”,在其下空白处右击“NewHadoopLocation”选项。在弹出的对话框中进行Hadoop的配置,如图4.18所示。其中“1”处配置“LocationName”,任意名称都可以,这里输入名称为“master”。“2”处的Host输入Hadoop集群中master节点的IP地址。“3”和“4”处分别设置通信端口号。“5”处设置通信的用户名,这里设置为“apache”。图4.18配置Hadooplocation

4. 3案例:使用MapReduce实现反向索引4.3.1Windows下Eclipse远程开发环境搭建(2)配置Eclipse配置hostname的解析。为了使Windows系统与Hadoop集群顺利通信,还需要配置hostname的解析。在Windows下打开“C:\Windows\System32\drivers\etc\hosts”文件,添加如下内容:00master01slave102slave2

4. 3案例:使用MapReduce实现反向索引4.3.1Windows下Eclipse远程开发环境搭建(2)配置Eclipse进行通信测试启动Hadoop集群后,打开Eclipse,单击左侧的DFSLocations,如果连接成功,在ProjectExplorer的DFSLocations下会看到master下的HDFS集群中的文件,如图4.19所示。此时,可以直接在Eclipse中操作HDFS集群上的文件。图4.19DFSLocations展现

4. 2MapReduce运行机制4.3.2 使用MapReduce实现反向索引(1)案例说明反向索引也称为倒排索引或反向档案,用来存储某个单词在一个文档或一组文档中的存储位置的映射。它是文档检索系统中最常用的数据结构,搜索引擎就是利用反向索引来进行搜索的。通过反向索引可以快速地通过文章内的单词反向检索获取包含这个单词的文档标识列表,从而完成巨大文件的快速搜索。

4. 2MapReduce运行机制4.3.2 使用MapReduce实现反向索引(1)案例说明假设有两个文件a.txt和b.txt,其内容如下:a.txt：helloworldhellospring

b.txt:helloapachehellohadoop通过反向索引后，得到的结果为：apacheb.txt:1hadoopb.txt:1helloa.txt:2;b.txt:2springa.txt:1worlda.txt:1

4. 2MapReduce运行机制4.3.2 使用MapReduce实现反向索引(1)案例说明整个实现过程:map函数解析输入的文档,输出一系列<单词,文档标识及单词次数>的键值对。MapReduce将键值对按照单词进行排序,然后局部合并相同单词的文档,形成一批“<单词,list(文档标识及单词次数)>”。reduce函数再将这些键值对进行进一步地排序、合并,最后输出反向索引结果。MapReduce反向索引实现原理示意图如图4.20所示。图4.20MapReduce反向索引实现原理示意图

4. 2MapReduce运行机制4.3.2 使用MapReduce实现反向索引(2)数据集本案例的数据集为“InverseData.zip”,包含三个英文文件,first.txt、second.txt和third.txt。首先,需要将这三个文件上传到HDFS的“/mrtestlin/”目录下。

文件上传到DFS的