基于Hadoop生态系统的大数据解决方案TP03

第三章

MapReduce实现分布式运算预习检查说出MapReduce的作用简述MapReduce的架构设计简述MapReduce的编程模型编写一个MapReduce程序应包括哪三个部分？简述MapReduce作业的执行流程集中测试2/39回顾与作业点评HDFS的作用是什么？管理HDFS文件有几种方式？编程实现HDFS文件操作有几种方式？分别是什么？区别是什么？点评作业的提交情况和共性问题回顾作业点评3/39本章任务统计音乐播放记录文件中的单曲播放次数4/39本章目标理解MapReduce计算模型掌握MapReduce应用开发掌握MapReduce程序在集群上的运行5/39生活中的MapReduce提问现在需要3个人一起工作,快速对这堆报纸分类，怎么做?6/39MapReduce基础5-1并行计算将大型而复杂的任务分解为许多小型任务同时计算，以提高计算机的处理能力MapReduce是实现了并行计算的编程模型为用户提供接口，屏蔽并行计算的细节map：映射，负责将任务分解成多个子任务reduce：归约，负责把分解后多任务的处理结果进行汇总7/39MapReduce基础5-2四个基本概念JobMapReduce程序编写完成后需要配置成一个MapReduce作业Task作业中的任务，分为两类：MapTask和ReduceTaskJobTracker负责创建MapTask或ReduceTaskTaskTracker负责执行MapTask或ReduceTask8/39MapReduce基础5-3架构设计MapReduceV1Master：JobTracker，初始化作业、创建调度子任务Slave：TaskTracker，执行任务MapReduceV2：YARN9/39MapReduce基础5-4MapReduce执行数据流inputsplitmapshufflereduce10/39MapReduce基础5-5MapReduce作业的工作原理MapReduce编程模型Mapper的实现Reducer的实现Job的配置11/39MapReduce应用开发基础4-1org.apache.hadoop.io.Writable接口Hadoop重新实现序列化机制Hadoop数据类型都实现该接口Hadoop基本数据类型IntWritable：整型数LongWritable：长整型数FloatWritable：浮点数DoubleWritable：双字节数BooleanWritable：布尔型ByteWritable：单字节，对应Java中byte类型BytesWritable：字节数组，对应Java中byte[]12/39MapReduce应用开发基础4-2Hadoop其它常用数据类型Text：使用UTF8格式存储的文本，对String类型的重写ObjectWritable：是一种对多类型的封装可以用于Java的基本类型，String等如同Java的Object类NullWritable：是一个占位符，序列化长度为零当<key,value>中的key或value为空时使用ArrayWritable：针对数组的数据类型MapWritable：对java.util.Map的实现演示示例1：使用Hadoop数据类型13/39MapReduce应用开发基础4-3MapReduce输入格式TextInputFormat：默认的数据输入格式KeyValueTextInputFormat：读取文本键值对SequenceFileInputFormat：读取二进制键值对hellohadoophellohiworld读取<0,hellohadoop><12,hellohiworld>hellohadoophellohiworld读取<hello,hadoop><hello,hiworld>LongWritableTextTextText14/39MapReduce应用开发基础4-4MapReduce输出格式

TextOutputFormat：默认的数据输出格式，以键值对的方式把数据写入一个文本文件SequenceFileOutputFormat：输出二进制键值对的顺序文件1hello2hadoop写入<1,hello><2,hadoop>LongWritableText15/39编写MapReduce程序实现Mapper继承org.apache.hadoop.mapreduce.Mapper实现Reducer继承org.apache.hadoop.mapreduce.Reducer创建并配置Job实例化org.apache.hadoop.mapreduce.Job实现步骤16/39步骤一：实现Mapper继承Mapper类并重写map()方法publicclassMapper<KEYIN,VALUEIN,KEYOUT,VALUEOUT>{ protectedvoidmap(KEYINkey,VALUEINvalue,Contextcontext){

context.write((KEYOUT)key,(VALUEOUT)value); }...//省略其它方法定义}依次为：输入键、输入值、输出键、输出值的类型MapReduce任务所在的上下文，是map任务和reduce任务间的桥梁Hellohadoop1Hellohadoop2Hellohadoop3split<0,Hellohadoop1><13,Hellohadoop2><26,Hellohadoop3>输入分片作为map()的输入publicvoidmap(Objectkey,Textvalue,Contextcontext)

{

IntWritablevalueOut=newIntWritable(1); StringTokenizertoken=newStringTokenizer(value.toString()); while(token.hasMoreTokens()){

TextkeyOut=newText(token.nextToken());

context.write(keyOut,valueOut); }}示例<Hello,1><hadoop1,1><Hello,1><hadoop2,1><Hello,1><hadoop3,1>17/39步骤二：实现Reducer继承Reducer并重写reduce()方法publicclassReducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT>{ protectedvoidreduce(KEYINkey,Iterable<VALUEIN>values,Contextcontext){ for(VALUEINvalue:values){ context.write((KEYOUT)key,(VALUEOUT)value);} }...//省略其它方法定义}<Hello,[1,1,1]><hadoop1,[1]><hadoop2,[1]><hadoop3,[1]><Hello,1><hadoop1,1><Hello,1><hadoop2,1><Hello,1><hadoop3,1>mapshufflepublicvoidreduce(Textkey,Iterable<IntWritable>values, Contextcontext){

intsum=0;

for(IntWritableval:values){

sum+=val.get();//将每个单词的每次统计求和

} context.write(key,newIntWritable(sum));}示例<Hello,3><hadoop1,1><hadoop2,1><hadoop3,1>18/39步骤三：创建MapReduce作业实例化、配置Job类Jobjob=newJob(conf,"wordcount");//创建作业job.setJarByClass(WordCount.class);//通过class查找job的jar文件job.setMapperClass(TokenizerMapper.class);//指定Mapperjob.setReducerClass(IntSumReducer.class);//指定Reducerjob.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);//设置输入文件路径或目录FileInputFormat.addInputPath(job,newPath(“hdfs://node1:9000/input/file”));//设置输出文件目录FileOutputFormat.setOutputPath(job,newPath(“hdfs://node1:9000/output/wordcount”));System.exit(job.waitForCompletion(true)?0:1);//提交作业同时指定Mapper和Reducer的键/值输出类型;Mapper的键/值输入类型由输入格式决定；Reducer的键/值输入类型由Mapper的输出类型决定19/39执行MapReduce程序运行WordCount打包jar文件运行命令实现步骤演示示例2：实现并运行WordCounthadoopjarwordcount.jarcn.mr.WordCount[Inputpath][OutputPath]示例20/39学员操作——统计单曲播放次数需求说明要求能够对于所有的音乐播放文件中的每一单曲进行播放次数统计实现思路新建MapReduce项目并配置属性core-site.xml：fs.defaultFSmapred-site.xml：实现map处理实现reduce处理配置作业运行作业，并查看输出结果指导完成时间：20分钟21/39常见问题及解决办法代码规范问题调试技巧共性问题集中讲解共性问题集中讲解22/39MapReduce作业执行流程作业的提交任务程序代码map()reduce()作业配置Mapper和Reducer的形参类型Mapper和Reducer的输入输出格式输入输出文件路径作业的初始化分配任务执行任务进度的更新作业的完成作业提交后，由此往下由框架自动执行23/39shuffle和排序4-1Mapper输出是如何转化为Reducer输入的？<Hello,[1,1,1]><hadoop1,[1]><hadoop2,[1]><hadoop3,[1]><Hello,1><hadoop1,1><Hello,1><hadoop2,1><Hello,1><hadoop3,1>shuffle问题Mapper输出Reducer输入经验虽然大部分情况下，我们无需对该阶段进行编程。但shuffle是MapReduce的核心，shuffle过程的性能与整个MapReduce的性能直接相关!24/39shuffle和排序4-2shuffle（混洗）的三个过程分区溢写排序spill0.outspill1.out...合并.out输出文件按分区合并所有Map任务的输出文件，同时保证在该分区内按键有序25/39shuffle和排序4-3既然shuffle可以保证Reducer的输入在各分区内是按键有序的，那么能不能保证所有分区之间有序呢？又该如何保证全局有序呢？问题分析<1,[...]><75,[...]><100,[...]>Reducer1的输入<50,[...]><85,[...]><150,[...]>Reducer2的输入使分区间有序<1,[...]><50,[...]><75,[...]>Reducer1的输入<85,[...]><100,[...]><150,[...]>Reducer2的输入26/39shuffle和排序4-4使分区间有序自定义Partitioner重写getPartition()方法关键：如何确定分区的边界为作业配置Partitionerjob.setPartitionerClass(Class)实现降序自定义Comparator重写compare()方法应针对不同的key类型实现相应的Comparator两个compare()方法都应重写为作业配置SortComparatorClassjob.setSortComparatorClass(Class)演示示例3：实现音乐播放记录降序统计示例27/39学员操作——统计并排序需求说明对练习1的统计结果进行排序自定义分区处理：key为[song1-song4]划分到分区0，其余划分到分区1对统计结果按播放次数降序输出实现思路创建2个作业，首先作业1完成统计，然后作业2完成排序自定义分区类，按单曲名称分别划分到不同分区，并指定mapreduce.job.reduces值为2作业1可使用SequenceFileOutputFormat作为输出格式作业2相应接收作业1的输出，其Mapper处理使用内置的InverseMapper实现key、value对调，并限定reduce个数为1重写IntWritable.Comparator实现降序指导完成时间：20分钟28/39常见问题及解决办法代码规范问题调试技巧共性问题集中讲解共性问题集中讲解29/39MapReduceV1的局限性JobTracker单点瓶颈TaskTracker端作业分配信息过于简单作业延迟高编程框架不够灵活30/39YARN3-1对MapReduceV1的改进JobTracker：分离集群资源管理和作业管理TaskTracker：演化为节点管理器NodeManager架构设计资源管理器节点管理器应用主体资源容器31/39YARN3-2配置文件yarn-evn.sh需指定JDK路径mapred-site.xml指定为yarnyarn-site.xml主要配置属性yarn.resourcemanager.addressyarn.resourcemanager.scheduler.addressyarn.resourcemanager.resource-tracker.addressyarn.resourcemanager.admin.addressyarn.resourcemanager.webapp.address需单独启动sbin/start-yarn.sh32/39YARN3-3MapReduce作业执行流程优势提高了集群的扩展性扩展了YARN的适用范围提高了集群资源的利用率33/39学员操作——查看YARN运行状态需求说明使用Web页面查看YARN中的作业查看正在运行的作业的状态、