分布式云计算dr

《云计算：从基础架构到

最佳实践》第12章分布式云计算

（时间：1次课，2学时）内容提要本章以Hadoop的MapReduce框架为重点，介绍MapReduce的起源、架构、原理，以及如何搭建多节点Hadoop计算集群，最后介绍MapReduce的编程方式，并通过示例演示如何开发MapReduce应用程序通过本章学习，读者应能够理解MapReduce原理，掌握ＭapReduce的配置与管理本章要点MapReduce简介MapReduce的原理MapReduce的编程框架Hadoop多集群部署MapReduce编程示例12分布式云计算12.1MapReduce简介12.2MapReduce集群部署12.3MapReduce开发接口MapReduceMapReduce是一种分布式计算编程模型，适用于万亿字节(TB)数量级的数据处理MapReduce通过封装一组通用的编程开发库，使得开发人员无须关注分布式细节，就可以将自己的程序部署于分布式环境MapReduce原理MapReduceMapReduce由Map(映射)、Reduce(化简)两个阶段组成对等待处理的大数据进行分块，根据任务节点数，分成M块(假设有M个计算节点)在各个节点运行Map函数，Map函数的作用是在本地节点上对数据分块进行运算，生成类似于<key,value>的数据集合，记作：list<<key,value>>在各个节点运行Reduce函数，Reduce的输入是各个节点Map函数的输出，对于M个(list<<key,value>>)输入源，每个Reduce节点只处理与本节点Map输出的数据(list<key,value>)相同键值的数据，处理后的节点仍以list<key,value>的格式存放于本地将各个节点上的Reduce输出串接起来，得到的结果就是本次MapReduce任务所期望的结果Hadoop的MapReduceHadoop完整地实现了MapReduce框架，当前大家所使用的MapReduce是一个易于使用的分布式开发框架，基于它编写的程序可以在数千个集群节点上并行运行HadoopMapReduce集群由一个MasterJobTracker与若干个SlaveTaskTracker组成Master节点的作用是将一个任务调度至各Slave节点上，将大规模数据集分片分配给各个Slave节点，监测各Slave节点的任务执行情况在容错方面，Master节点定期与各Slave节点通信，获取其运行状态，对一定时间间隔内没有响应的Slave节点进行隔离，并将它的其任务重新指派给其他Slave节点Slave节点的作用是接受Master的控制命令，执行具体的任务Hadoop的MapReduce工作流程五个阶段文件输入Map作业临时文件生成Reduce作业结果文件生成Hadoop的MapReduce工作流程步骤将用户程序分别复制到集群内的Master与worker(就是前面说的Slave)节点中，同时把输入文件分成若干份，图12-2中分成五份(split0～split4)Master节点负责为空闲worker节点分配任务，如：有的worker节点被分配Map作业，有的被分配Reduce作业，为了减少数据内部通信，通常worker节点在执行完Map作业后，会被继续分配Reduce作业在分配到Map作业的worker节点读取对应的数据块，并从中抽取出<key,value>对，产生的键值列表被存放至内存中对于已计算完的<key,value>对会被定期写入本地磁盘，在写入过程中会将<key,value>键值对再分N个子区，并将每个<key,value>键值对所属的子区、worker节点位置上报给Master节点Hadoop的MapReduce工作流程步骤Master通知执行Reduce作业的worker读取指定的Map作业产生的键值对，通常Reduce作业节点先会对远程读取到的键值对进行统一排序，再交给Reduce函数处理当Reduce作业节点完成处理后，将结果写入本地磁盘文件，当所有的Map与Reduce作业完成后，向用户端返回执行结果MapReduce控制与命令MapReduce经常使用的控制命令运行JobTracker节点运行TaskTracker节点对已运行的MapReduce任务进行操作查看运行任务队列MapReduce控制与命令运行MapReduce任务通过jps命令可以查看JobTracker启动hadoopjobtrackerTaskTracker启动hadooptasktracker[root@master/]#hadoopjobtracker12/10/1510:52:09INFOmapred.JobTracker:STARTUP_MSG:[root@master/]#hadooptasktracker12/10/1510:54:38INFOmapred.TaskTracker:STARTUP_MSG:[root@master/]#jps29757Jps29656TaskTracker29148DataNode29033NameNode29346JobTracker29259SecondaryNameNode7478 sterMapReduce控制与命令查看队列通过hadoopqueue命令可以查看运行的任务队列及任务的具体信息[root@master/]#hadoopqueue-listQueueName:defaultSchedulingInfo:N/A[root@master/]#hadoopqueue-infodefaultQueueName:defaultSchedulingInfo:N/A[root@master/]#hadoopqueue-infodefault-showJobsQueueName:defaultSchedulingInfo:N/AJobListJobIdStateStartTimeUserNamePrioritySchedulingInfoMapReduce控制与命令管理MapReduce任务针对MapReduce任务管理，Hadoop提供了job命令，这个命令的功能包括查看当前运行任务列表、查看每个任务的运行信息、设置任务优先级、提交任务、关闭任务[root@master/]#hadoopjobUsage:JobClient<command><args>[-submit<job-file>][-status<job-id>][-counter<job-id><group-name><counter-name>][-kill<job-id>][-set-priority<job-id><priority>].Validvaluesforprioritiesare:

VERY_HIGHHIGHNORMALLOWVERY_LOW[-events<job-id><from-event-#><#-of-events>][-history<jobOutputDir>][-list[all]][-list-active-trackers][-list-blacklisted-trackers][-list-attempt-ids<job-id><task-type><task-state>]

[-kill-task<task-id>][-fail-task<task-id>]MapReduce控制与命令管理MapReduce任务-submit，提交任务，后面指定任务可执行文件名-status，查看状态，打印指定任务的map与reduce过程完成率-counter，打印计数器的值-kill，杀死一个指定的作业-set-priority，设置任务优化级，可选的值包括VERY_HIGHHIGH(高)、NORMALLOW(正常)、VERY_LOW(低)-events，输出指定范围内指定任务接受到的事件列表与细节-history，打印指定任务的全部作业细节-list-blacklisted-trackers，显示当前被禁用的节点列表-kill-task，杀死指定的任务-fail-task，使任务失败MapReduce控制与命令管理MapReduce任务-list，显示作业，其中listall显示全部作业，list只显示正在完成的作业[root@master/]#hadoopjob-list0jobscurrentlyrunningJobIdStateStartTimeUserNamePrioritySchedulingInfo[root@master/]#hadoopjob-listall0jobssubmittedStatesare:Running:1Succeded:2Failed:3Prep:4JobIdStateStartTimeUserNamePrioritySchedulingInfo[-fail-task<task-id>]MapReduce控制与命令管理MapReduce任务-list-active-trackers，显示当前存活的节点名列表[root@master/]#hadoopjob-list-active-trackerstracker_master.lb.rde.testlinux:localhost.localdomain/127.0.0.1:21048MapReduce控制与命令管理MapReduce任务-list-attempt-ids，显示满足<作业、任务、状态>列表如显示作业1中所有状态为running的map任务列表[root@master/]#hadoopjob-list-attempt-ids1maprunning12分布式云计算12.1MapReduce简介12.2MapReduce集群部署12.3MapReduce开发接口MapReduce集群部署提供集群化部署Hadoop集群设计一个三节点的Hadoop集群，由一个节点承担名字节点、数据节点、任务节点、调度节点等任务；另外两个只承担数据节点与任务节点主要步骤（详细过程见实验大纲）准备物理与操作系统环境配置网络地址、域名三台机器之间SSH免登录JavaSDK安装与配置Hadoop安装与配置Hadoop格式化与服务启动查看效果12分布式云计算12.1MapReduce简介12.2MapReduce集群部署12.3MapReduce开发接口MapReduce开发接口MapReduce编程框架用户向Hadoop集群提交一个MapReduce作业，Hadoop集群会将这个作业分解成一系列的map与reduce任务，分发至各TaskTracker节点执行，并对运行结果与作业进度进行监测两种开发接口JavaAPI开发MapReduce应用streaming机制利用Shell、C/C++编写Map与Reduce任务MapReduce开发接口MapReduce编程过程map函数编写围绕map函数，需要定义信息的输入格式，map函数只处理<key,value>，这里需要指定key与value的数据类型，其中value可以是复杂的数据对象reduce函数编写围绕reduce函数，需要定义输出格式，输出的格式也是<key,value>模式作业配置设置作业的属性、名称、指定输入与输出的格式，指定map与reduce可执行代码的位置，指定输入与输出信息的位置等JavaAPI介绍与示例Map方法map以一个Java类的形式存在，它需要继承Mapper接口，并实现Mapper中的map方法。voidmap(K1key,

V1value,

OutputCollector<K2,V2>output,

Reporterreporter

)throwsIOExceptionK1key与V1value是指map方法要处理的输入，其中K1与V1为数据类型，可以选择的数据类型包括Text、DoubleWritable、IntWritable、LongWritable等OutputCollector<K2,V2>output，用于指定map方法执行后生成的结果，这里的K2,V2分别对应结果的键与值JavaAPI介绍与示例Map方法演示publicstaticclassMapextendsMapper<K1,V1,K2,V2>{publicvoidmap(K1key,V1value,OutputCollector<K2,V2>output,

Reporterreporter)throwsIOException,InterruptedException{

}}JavaAPI介绍与示例Reduce方法与Map方法相类似，Reduce同样以类的形式存在，它需要继承Reducer接口，并实现Reduce方法voidreduce(K2key,Iterator<V2>values,OutputCollector<K3,V3>output,Reporterreporter)throwsIOExceptionK2key与Iterator<V2>values，对应Map过程产生的输出，这里的values是一个列表，这是由于Map过程后对于同一个键(key)，可能会有多个输出值。values里的值经由Hadoop对Map输出通过归并、排序后重新组织。OutputCollector<K3,V3>output，对应于Reduce的输出，也是最终的输出。最终的输出格式需要用户自定义JavaAPI介绍与示例Reduce方法演示publicstaticclassReduceextendsReducer<K2,V2,K3,V3>{publicvoidreduce(K2key,Iterator<V2>values,OutputCollector<K3,V3>

output,Reporterreporter)throwsIOException,InterruptedException{

}}JavaAPI介绍与示例MapReduce作业配置对执行的作业进行组织与配置，由JobConf类来配置，由JobClient运行初始化配置任务，指定任务主类名，配置任务名称。newJobConf(newConfiguration(),Job.class);JobConf.setJobName("myjob");设置Map与Reduce执行类。JobConf.setMapperClass(Map.class);JobConf.setCombinerClass(Reduce.class);JobConf.setReducerClass(Reduce.class);设置输入与输出路径。FileInputFormat.setInputPaths(conf,newPath(args[0]));FileOutputFormat.setOutputPath(conf,newPath(args[1]));设定输入与输出格式。JobConf.setInputFormat(SequenceFileInputFormat.class);JobConf.setOutputFormat(SequenceFileOutputFormat.class);JobConf.setOutputValueClass(IntWritable.class);JobConf.setOutputKeyClass(Text.class);JavaAPI介绍与示例示例：统计访问量的Reduce任务互联网网站每天都有大量用户访问，因此会产生大量的访问日志，如何分析用户请求，计算出每天页面访问总量，单个页面的访问量、访问响应情况等是一项基础性工作。这里设计一个MapReduce任务用于分析并处理访问日志，生成需要的统计数据时间访问页面名称返回值20120909a.htm20020120909b.htm20020120909c.htm40420120909d.htm20020120909a.htm20020120909a.htm20020120909d.htm20020120909d.htm20020120909a.htm200JavaAPI介绍与示例示例：统计访问量的Reduce任务编写这个MapReduce任务可以等同于查找每个单词出现的次数如果单词是时间字段，则代表每天页面的访问量如果是访问页面名称，则代表当天这个页面被访问的次数如果是请求响应码200，则可以标记当天访问成功的次数若是404则代表当天访问失败的次数处理步骤Map类：将文件以空格、换行符为间隔，抽取成<key,1>格式，如<20120909,1>Reduce类：对一系列<key,1>进行整合与查找，如果key值相同，则相加后面的value值，最后归约生成的结果就是所有key的统计列表AccessStat类：用于配置任务JavaAPI介绍与示例Map类importjava.io.*;importjava.util.*;importorg.apache.hadoop.io.*;importorg.apache.hadoop.mapred.*;importorg.apache.hadoop.util.*;importorg.apache.hadoop.fs.*;

publicclassMapextendsMapper<LongWritable,Text,Text,IntWritable>{publicvoidmap(LongWritablekey,Textvalue,OutputCollector<Text,

IntWritable>output,Reporterreporter)throwsIOException{try{StringstrMessage=value.toString();TextstrKey=newText();StringTokenizertokenizer=newStringTokenizer(strMessage);while(tokenizer.hasMoreTokens()){

strKey.set(tokenizer.nextToken());

output.collect(strKey,newIntWritable(1));}}catch(Exceptione){e.printStackTrace();}}}JavaAPI介绍与示例Reduce类importjava.io.*;importjava.util.*;importorg.apache.hadoop.io.*;importorg.apache.hadoop.mapred.*;

publicclassReduceextendsReducer<Text,IntWritable,Text,IntWritable>{publicvoidreduce(Textkey,Iterable<IntWritable>values,OutputCollector<Text,IntWritable>output,Reporterreporter)throwsIOException{try{intsum=0;while(values.hasNext()){sum+=values.next().get();}output.collect(key,newIntWritable(sum));

}catch(Exceptione){e.printStackTrace();

}}}

JavaAPI介绍与示例AccessStat类importjava.io.*;importjava.util.*;importorg.apache.hadoop.conf.*;importorg.apache.hadoop.io.*;importorg.apache.hadoop.mapred.*;importorg.apache.hadoop.util.*;importorg.apache.hadoop.fs.*;

publicclassAccessStat{publicstaticvoidmain(String[]args)throwsException{try{JobConfconf=newJobConf(AccessStat.class);conf.setOutputValueClass(IntWritable.class);conf.setOutputKeyClass(Text.class);conf.setMapperClass(Map.class);conf.setCombinerClass(Reduce.class);conf.setReducerClass(Reduce.class);conf.setInputFormat(TextInputFormat.class);conf.setOutputFormat(TextOutputFormat.class);conf.setJobName("AccessStat");FileInputFormat.setInputPaths(conf,newPath(args[0]));FileOutputFormat.setOutputPath(conf,newPath(args[1]));JobClient.runJob(conf);}catch(Exceptione){e.printStackTrace();}}}JavaAPI介绍与示例程序编译通过javac编译三个Java文件[root@masterMapReduceEx01]#javacMap.java–classpath/usr/local/hadoop/hadoop-0.20.2-core.jar-source5[root@masterMapReduceEx01]#javacReduce.java–classpath/usr/local/hadoop/hadoop-0.20.2-core.jar-source5[root@masterMapReduceEx01]#javacAccessStat.java–classpath/usr/local/hadoop/hadoop-0.20.2-core.jar-source5JavaAPI介绍与示例程序打包通过jar对所有类文件进行打包[root@masterMapReduceEx01]#vimanifest.mfManifest-Version:1.0Class-Path:/usr/local/hadoop/hadoop-0.20.2-core.jarMain-Class:AccessStat[root@masterMapReduceEx01]#jarcvfAccessStat*.classadding:META-INF/(in=0)(out=0)(stored0%)adding:META-INF/MANIFEST.MF(in=56)(out=56)(stored0%)adding:AccessStat.class(in=1699)(out=714)(deflated57%)adding:Map.class(in=1699)(out=714)(deflated57%)adding:Reduce.class(in=1699)(out=714)(deflated57%)Total:------(in=5141)(out=2720)(deflated47%)JavaAPI介绍与示例程序运行在Hadoop集群中创建目录MapReduce01，创建输入文件[input.txt]，并上传至Hadoop集群中[root@masterMapReduceEx01]#hadoopfs-mkdir/home/HadoopData/MapReduce01[root@masterMapReduceEx01]#viinput.txt20120909a.htm20020120909b.htm20020120909c.htm40420120909d.htm20020120909a.htm20020120909a.htm20020120909d.htm20020120909d.htm20020120909a.htm200JavaAPI介绍与示例程序运行在Hadoop集群中创建目录MapReduce01，创建输入文件[input.txt]，并上传至Hadoop集群中[root@masterMapReduceEx01]#hadoopfs-copyFromLocalinput.txt/home/HadoopData/MapReduce01/input.txt[root@masterMapReduceEx01]#hadoopfs-lsr/home/HadoopData/MapReduce01-rw-r--r--1rootsupergroup1722012-10-1519:18/home/HadoopData/MapReduce01/input.txtJavaAPI介绍与示例程序运行通过Hadoop命令可以执行AccessStat.jar任务[root@masterMapReduceEx01]#hadoopjarAccessStat.jar/home/HadoopData/MapReduce01/input.txt/home/HadoopData/MapReduce01/output12/10/1519:28:31INFOmapred.FileInputFormat:Totalinputpathstoprocess:112/10/1519:28:31INFOstreaming.StreamJob:getLocalDirs():…

[root@master/]#hadoopfs-cat/home/HadoopData/MapReduce01/output/part-000009201209094a.htm1b.htm1c.htm3d.htm82001404streaming介绍与示例MapReduce支持多种编程语言，默认是Java语言，若要使用其他编程语言，需使用Hadoop的其他编程机制Hadoop提供了Streaming、Pipes、Pydoop编程机制，支持用其他语言编写Map与Reduce可执行文件或脚本，其中Pipes用于支持C++，Pydoop用于支持Python语言，Streaming可以支持其他任何语言Streaming是一个Java可执行文件它存放于$HADOOP_HOME/contrib/streaming/hadoop-0.20.2-streaming.jarstreaming介绍与示例查看streaming帮助[root@masterstreaming]#hadoopjarhadoop-0.20.2-streaming.jarUsage:$HADOOP_HOME/bin/hadoopjar$HADOOP_HOME/hadoop-streaming.jar[options]Options:-input<path>DFSinputfile(s)fortheMapstep-output<path>DFSoutputdirectoryfortheReducestep-mapper<cmd|JavaClassName>Thestreamingcommandtorun-combiner<JavaClassName>CombinerhastobeaJavaclass-reducer<cmd|JavaClassName>Thestreamingcommandtorun-file<file>File/dirtobeshippedintheJobjarfile-inputformatTextInputFormat(default)|SequenceFileAsTextInputFormat|JavaClassNameOptional.-outputformatTextOutputFormat(default)|JavaClassNameOptional.-partitionerJavaClassNameOptional.-numReduceTasks<num>Optional.-inputreader<spec>Optional.-cmdenv<n>=<v>Optional.Passenv.vartostreamingcommands-mapdebug<path>Optional.Torunthisscriptwhenamaptaskfails-reducedebug<path>Optional.Torunthisscriptwhenareducetaskfails-verboseStreaming介绍与示例streaming原理Streaming原理类似于JavaAPI机制，用户使用其他语言编写出Map与Reduce脚本或可执行文件在Streaming中配置Map与Reduce可执行程序或脚本路径、源数据输入、目标数据存放、输入的格式、输出的格式等信息Streaming会根据配置自动调用这些过程，并在调用时为Map提供输入，为Reduce提供Map产生的输出，并将最终的结果以指定格式存放在指定位置streaming的shell示例示例：统计访问量的Reduce任务MapMap脚本文件的作用是提取出input.txt中的每一个字符串，并顺次输出[root@masterMapReduceShell]#catmap#!/bin/shcat$1|awk'{print$1"\n"$2"\n"$3}'streaming的shell示例示例：统计访问量的Reduce任务测试Map脚本执行效果Map脚本文件的作用是提取出input.txt中的每一个字符串，并顺次输出[root@masterMapReduceShell]#./mapinput.txt20120909a.htm20020120909b.htm20020120909c.htm40420120909d.htm20020120909a.htm20020120909a.htm20020120909d.htm20020120909d.htm20020120909a.htm200streaming的shell示例示例：统计访问量的Reduce任务ReduceReduce脚本是通过sort进行归类排序后，通过uniq进行计数统计，并将结果输出[root@masterMapReduceShell]#catreduce#!/bin/shsort|uniq–cstreaming的shell示例示例：统计访问量的Reduce任务测试Reduce脚本执行效果组合Map与Reduce脚本运行效果[root@masterMapReduceShell]#./mapinput.txt|./reduce820092012090914044a.htm1b.htm1c.htm3d.htmstreaming的shell示例示例：统计访问量的Reduce任务通过streaming的hadoop集群执行效果[root@masterstreaming]#hadoopjarhadoop-0.20.2-streaming.jar-input/home/HadoopData/MapReduceShell/input.txt-output/home/HadoopData/MapReduceShell/input-mapper/home/MapReduceShell/map-reducer/home/MapReduceShell/reduce12/10/1610:56:44INFOmapred.FileInputFormat:Totalinputpathstoprocess:….[root@masterstreaming]#hadoopfs-cat/home/HadoopData/MapReduceShell/input/part-00000820092012090914044a.htm1b.htm1c.htm3 d.htmstreaming的C++示例示例：统计访问量的Reduce任务Map文件ma