Hadoop 3大数据部署与数据分析实战 课件02-03-Hadoop入门及实战-MapReduce

第五章MapReduce学习目标MapReduce概述——概念、思想和特点MapReduce编程模型（重点）MapReduce编程进阶MapReduce工作机制（重点）MapReduce编程案例第五章MapReduce5.1MapReduce概述曹聪称象+MapReduce概念MapReduce是Google开源的一项重要技术，它是一种简化并行计算的编程模型，用于进行大数据量的计算。1+2+3+4+5+6+7+8+9+10大任务1+2+3+45+6+78+9+10split10182755分解MapMapReduce思想归纳Reduce小任务“分而治之，百川归海”MapReduceMapReduce特点易于编程良好的扩展性高容错性擅长对PB级以上海量数据进行离线处理MapReduce不擅长的场景（1）实时计算MapReduce无法像MySQL一样，在毫秒或者秒级内返回结果。（2）流式计算

流式计算的输入数据是动态的，而MapReduce的输入数据集是静态的，不能动态变化。案例：京东大数据平台京东大数据平台技术架构来源：京东大数据技术白皮书案例：京东大数据平台京东大数据平台演进的几个重要节点分别是单集群规模达到1200台、3000台、5000台再到2018年的8000台。离线数据挖掘和分析——推荐系统、用户画像和针对性营销。你有过被大数据“杀熟”的经历吗？案例：流式计算——微博热搜实时计算框架——Spark（了解）ApacheSpark是一个开源集群运算框架，最初是由加州大学柏克莱分校AMPLab所开发。相对于Hadoop的MapReduce会在运行完工作后将中介资料存放到磁盘中，Spark使用了存储器内运算技术，能在资料尚未写入硬盘时即在存储器内分析运算。实时计算框架——Flink（了解）ApacheFlink是由Apache软件基金会开发的开源流处理框架，其核心是用Java和Scala编写的分布式流数据流引擎。Flink以数据并行和管道方式执行任意流数据程序，Flink的流水线运行时系统可以执行批处理和流处理程序。此外，Flink的运行时本身也支持迭代算法的执行。实时计算框架——Storm（了解）Storm是一个分布式计算框架，主要由Clojure编程语言编写。最初是由NathanMarz及其团队创建于BackType，该项目在被Twitter取得后开源。它使用用户创建的“管”和“螺栓”来定义信息源和操作来允许批量、分布式处理流式数据。第五章MapReduce5.2MapReduce编程模型MapReduce编程模型MapReduce由两个阶段组成：Map和Reduce。用户只需要实现map和reduce，即可完成分布式程序的设计。Map和Reduce阶段的输入输出都是一系列的键值对。MapReduce输入输出的键值对这里的键值对可以理解为Java中的List<Pair<k,v>>，由“键值对”组成的列表。#Java的Map的用法Map<String,String>map1=newHashMap<>();map1.put(“001”,”张三”);map1.put(“001”,”李四”);map1.get(“001”)#键值对列表List<Pair<k,v>>pairList=newArrayList<>();pairList.add(new

Pair(“001”,”张三”));pairList.add(new

Pair(“001”,”李四”));pairList输出什么？“李四”输出什么？[<“001”,”张三”>,<“001”,”李四”>]MapReduce编程模型输入一个大文件，通过分片之后，将其分为多个分片（inputsplit）每个文件分片由单独的机器去处理，这就是Map方法将各个机器计算的结果进行汇总并得到最终结果，这就是Reduce

方法MapReduce编程模型第1阶段：大文件分片第2阶段：Map处理

接收分片作为输入，输出处理结果。第3阶段：Reduce处理Map阶段处理结果作为输入，输出汇总结果1+2+3+4+5+6+7+8+9+10大任务1+2+3+45+6+78+9+10split10182755分解MapMapReduce思想归纳Reduce小任务MapReduce的Java编程要点最基础需要3个类，包括Mapper，Reducer

和带main方法的程序入口类。实例1.WordCount——分片计算一份文本文档当中单词出现的数量，先对文件进行分片agoodbeginningishalfthebattlewherethereisawillthereisawaywherethereisawillthereisawayagoodbeginningishalfthebattle实例1.WordCount——Map阶段输入<k1,v1>k1是行号，v1是每行内容<“#1”,”agoodbeginningishalfthebattle”><“#2”,”wherethereisawillthereisaway”>map()map()输出<k2,v2>k2是单词，v2是单词在每行出现次数<“a”,1><“half”,1><“battle”,1><“the”,1><“beginning”,1><“is”,1><“good”,1><“a”,1><“there”,1><“a”,1><“way”,1><“is”,1><“where”,1><“is”,1><“will”,1><“there”,1>实例1.WordCount——ShuffleMap阶段分区（partiton）Reduce阶段<k2,v2>k2是单词，v2是单词在每行出现次数合并（combine）——如果有就执行排序与溢写（sortandspill）<k3,v3><“the”,[1]><“there”,[1,1]><“way”,[1]><“where”,[1]><“will”,[1]>按照分区获取溢写出的文件，生成新的键值对Shuffle<“a”,1><“half”,1><“battle”,1><“the”,1><“beginning”,1><“is”,1><“good”,1><“a”,1><“there”,1><“a”,1><“way”,1><“is”,1><“where”,1><“is”,1><“will”,1><“there”,1><“a”,[1,1,1]><“half”,[1]><“battle”,[1]><“beginning”,[1]><“is”,[1,1,1]><“good”,[1]>实例1.WordCount——Reduce阶段输入reduce()输出<k3,v3>k3是单词，v3是合并相同k3的次数列表<“a”,3><“half”,1><“battle”,1><“the”,1><“beginning”,1><“way”,1><“is”,3><“where”,1><“there”,2><“will”,1><“good”,1><k4,v4>k4是单词，v4是单词出现总次数<“a”,[1,1,1]><“half”,[1]><“battle”,[1]><“beginning”,[1]><“is”,[1,1,1]><“good”,[1]><“the”,[1]><“there”,[1,1]><“way”,[1]><“where”,[1]><“will”,[1]>MapReduce大致工作流程（重点）MapReduce大致工作流程，主要分为八个步骤：(1)对输入文件进行切片规划。(2)启动相应数量的MapTask进程。(3)调用InputFormat中的RecordReader，读一行数据并封装为<k1,v1>,并将<k1,v1>传给map函数。(4)调用自定义的map函数计算，并输出<k2,v2>。(5)收集map的输出，进行分区和排序（shuffle),生成<k3,v3>。(6)ReduceTask任务启动，并从map端拉取数据。ReduceTask获取到输入数据<k3,v3>。(7)ReduceTask调用自定义的reduce函数进行计算，得到输出<k4,v4>。(8)调用Outputformat的RecordWriter将结果数据输出。MapReduce模型要点（重点）（1）任务Job=Map+Reduce（2）Map的输出是Reduce的输入（3）所有的输入和输出都是<Key,Value>形式

<k1,v1>是Map的输入，<k2,v2>是Map的输出

<k3,v3>是Reduce的输入，<k4,v4>是Reduce的输出（4）k2=k3，v3是相同键的值的集合。例如<k2,v2>存在<a,1>，<a,2>。会转换为<k3,v3>的<a,[1,2]>（5）所有的输入和输出的数据类型必须是hadoop的数据类型（实现Writable接口）

Integer->IntWritableLong->LongWritableString->Textnull->NullWritable（6）MapReduce处理的数据都是HDFS的数据（或HBase）MapReduce的Java编程要点最基础需要3个类，包括Mapper，Reducer

和带main方法的程序入口类。实例1——Wordcount代码Mapper继承org.apache.hadoop.mapreduce.Mapper，并实现map方法publicclassWordCountMapperextendsMapper<LongWritable,Text,Text,IntWritable>{ @Override protectedvoidmap(LongWritablekey1,Textv1,Contextcontext) throwsIOException,InterruptedException{ Stringdata=v1.toString();

//分词

String[]words=data.split("");

//输出k2v2 for(Stringw:words){ context.write(newText(w),newIntWritable(1)); } }}Reducer继承org.apache.hadoop.mapreduce.Reducer，并实现reduce方法publicclassWordCountReducerextendsReducer<Text,IntWritable,Text,IntWritable>{@Overrideprotectedvoidreduce(Textk3,Iterable<IntWritable>v3,Contextcontext)throwsIOException,InterruptedException{

//context是reduce的上下文

//对v3求和

inttotal=0; for(IntWritablev:v3){ total+=v.get(); }

//输出：k4单词v4频率

context.write(k3,newIntWritable(total)); }}Main作为调用程序的入口，需要实现main方法publicclassWordCountMain{publicstaticvoidmain(String[]args)throwsException{

//创建一个job和任务入口

Jobjob=Job.getInstance(newConfiguration());job.setJarByClass(WordCountMain.class);//main方法所在的class

//指定job的mapper和输出的类型<k2v2>job.setMapperClass(WordCountMapper.class);job.setMapOutputKeyClass(Text.class);//k2的类型

job.setMapOutputValueClass(IntWritable.class);//v2的类型

//指定job的reducer和输出的类型<k4v4>job.setReducerClass(WordCountReducer.class);job.setOutputKeyClass(Text.class);//k4的类型

job.setOutputValueClass(IntWritable.class);//v4的类型

//指定job的输入和输出

FileInputFormat.setInputPaths(job,newPath(args[0]));FileOutputFormat.setOutputPath(job,newPath(args[1]));

//执行jobjob.waitForCompletion(true);}}实例2——统计各个部门员工薪水总和实例2.统计各个部门员工薪水总和和WordCount类似，<k2,v2>使用<部门，薪水>案例解读实例2.统计各个部门员工薪水总和MapperpublicclassSalaryTotalMapperextendsMapper<LongWritable,Text,IntWritable,IntWritable>{ @Override protectedvoidmap(LongWritablekey1,Textvalue1,Contextcontext)throwsIOException,InterruptedException{

//数据：7654,MARTIN,SALESMAN,7698,1981/9/28,1250,1400,30 Stringdata=value1.toString();

//分词

String[]words=data.split(",");

//输出：k2部门号v2薪水

context.write(newIntWritable(Integer.parseInt(words[7])), newIntWritable(Integer.parseInt(words[5])));//薪水

}}实例2.统计各个部门员工薪水总和ReducerpublicclassSalaryTotalReducerextendsReducer<IntWritable,IntWritable,IntWritable,IntWritable>{@Overrideprotectedvoidreduce(IntWritablek3,Iterable<IntWritable>v3,Contextcontext)throwsIOException,InterruptedException{

//对v3求和，得到该部门的工资总额

inttotal=0; for(IntWritablev:v3){ total+=v.get(); }

//输出：部门号总额

context.write(k3,newIntWritable(total)); }}实例2.统计各个部门员工薪水总和Mainpublicstaticvoidmain(String[]args)throwsException{

//创建一个job和任务入口

Jobjob=Job.getInstance(newConfiguration()); job.setJarByClass(WordCountMain.class);//main方法所在的class

//指定job的mapper和输出的类型<k2v2> job.setMapperClass(WordCountMapper.class); job.setMapOutputKeyClass(Text.class);//k2的类型

job.setMapOutputValueClass(IntWritable.class);//v2的类型

//指定job的reducer和输出的类型<k4v4> job.setReducerClass(WordCountReducer.class); job.setOutputKeyClass(Text.class);//k4的类型

job.setOutputValueClass(IntWritable.class);//v4的类型

//指定job的输入和输出

FileInputFormat.setInputPaths(job,newPath(args[0])); FileOutputFormat.setOutputPath(job,newPath(args[1]));

//执行job job.waitForCompletion(true);}实例3——统计各个部门员工薪水总和（使用序列化Key或Value）实例3.统计各个部门员工薪水总和（使用序列化Key或Value）序列化是指Java对象

传输数据流<k2,v2>使用<部门，Employee类>自定义的Key或Value要继承org.apache.hadoop.io.Writable接口MapReduce大致工作流程（重点回顾）MapReduce大致工作流程，主要分为八个步骤：(1)对输入文件进行切片规划。(2)启动相应数量的MapTask进程。(3)调用InputFormat中的RecordReader，读一行数据并封装为<k1,v1>,并将<k1,v1>传给map函数。(4)调用自定义的map函数计算，并输出<k2,v2>。(5)收集map的输出，进行分区和排序（shuffle)。(6)ReduceTask任务启动，并从map端拉取数据。ReduceTask获取到输入数据<k3,v3>。(7)ReduceTask调用自定义的reduce函数进行计算，得到输出<k4,v4>。(8)调用Outputformat的RecordWriter将结果数据输出。MapperReducer实例2.统计各个部门员工薪水总和第一步——看最终结果可以确定<k4,v4><k4,v4><部门编号，薪水总和>程序设计关键在于如何分解任务，确定<k,v>K2，K3，K4一般很少会变化，大多数场景都保持一致。实例2.统计各个部门员工薪水总和第二步——看输入，如何分解任务横向切分实例2.统计各个部门员工薪水总和第三步在Map阶段取到需要的数据实例2.统计各个部门员工薪水总和第四步确认如何处理数据<key,value><部门号，薪水A+薪水B+...>Sum累加实例3.统计各个部门员工薪水总和（使用序列化Key或Value）自定义的Key或Value要继承org.apache.hadoop.io.Writable接口publicinterfaceWritable{/***序列化写入对象**/voidwrite(DataOutputout)throwsIOException;/***反序列化读取对象**/voidreadFields(DataInputin)throwsIOException;}实例3.统计各个部门员工薪水总和序列化对象publicclassEmployeeimplementsWritable{privateintempno;privateStringename;privateStringjob;...@OverridepublicvoidreadFields(DataInputinput)throwsIOException{//反序列化this.empno=input.readInt();...}@Overridepublicvoidwrite(DataOutputoutput)throwsIOException{//序列化output.writeInt(this.empno);output.writeUTF(this.ename);...}实例3.统计各个部门员工薪水总和MapperpublicclassSalaryTotalMapperextendsMapper<LongWritable,Text,IntWritable,Employee>{ @Override protectedvoidmap(LongWritablekey1,Textvalue1,Contextcontext) throwsIOException,InterruptedException{

//数据：7654,MARTIN,SALESMAN,7698,1981/9/28,1250,1400,30 Stringdata=value1.toString();

//分词

String[]words=data.split(",");

//创建员工对象

Employeee=newEmployee();

//设置员工的属性

e.setEmpno(Integer.parseInt(words[0]));//员工号

e.setEname(words[1]);//姓名 ...

//输出：k2部门号v2员工对象

context.write(newIntWritable(e.getDeptno()),e);//员工的部门号，员工对象

}}实例3.统计各个部门员工薪水总和ReducerpublicclassSalaryTotalReducerextendsReducer<IntWritable,Employee,IntWritable,IntWritable>{@Overrideprotectedvoidreduce(IntWritablek3,Iterable<Employee>v3,Contextcontext)throwsIOException,InterruptedException{//取出v3中的每个员工数据，进行工资求和inttotal=0;for(Employeee:v3){ total=total+e.getSal();}//输出context.write(k3,newIntWritable(total));}}实例3.统计各个部门员工薪水总和MainpublicclassSalaryTotalMain{ publicstaticvoidmain(String[]args)throwsException{

//创建一个job Jobjob=Job.getInstance(newConfiguration()); job.setJarByClass(SalaryTotalMain.class);

//指定job的mapper和输出的类型k2v2 job.setMapperClass(SalaryTotalMapper.class); job.setMapOutputKeyClass(IntWritable.class); job.setMapOutputValueClass(Employee.class);

//指定job的reducer和输出的类型k4v4 job.setReducerClass(SalaryTotalReducer.class); job.setOutputKeyClass(IntWritable.class); job.setOutputValueClass(IntWritable.class);

//指定job的输入和输出的路径

FileInputFormat.setInputPaths(job,newPath(args[0])); FileOutputFormat.setOutputPath(job,newPath(args[1]));

//执行任务

job.waitForCompletion(true); }}本章总结MapReduce概述MapReduce特点、不擅长场景MapReduce编程模型分片、Mapper、Reducer、Main。MapReduce工作流程和模型要点第五章MapReduce5.3MapReduce编程进阶MapReduce编程进阶MapReduce的输入格式(InputFormat)MapReduce的输出格式(OutputFormat)分区(Partition)合并(Combiner)MapReduce的输入格式InputFormat提供以下两个功能数据切分：按照某个策略将输入数据切分成若干个SplitInput，以便确定Mapper个数以及对应的SplitInput个数，这样一来，SplitInput的个数与Mapper个数总是相同的。为Mapper提供输入数据：读取给定的SplitInput的数据，解析成一个个的key/value对，供Mapper使用。MapReduce的输入格式agoodbeginningishalfthebattlewherethereisawillthereisawaywherethereisawillthereisawayagoodbeginningishalfthebattleSplit（切分）InputSplitInputSplitRecordReader（读取）<#1,”agoodbeginningishalfthebattle”><#2,”wherethereisawillthereisaway”>File（文件）MapReduce的输出格式OutputFormat主要用于描述输出数据的格式，通过RecordWriter能够将用户提供的key/value对写入特定格式的文件中TextOutputFormat调用toString()方法把它们转换为字符串NullWritable来省略输出的key或valueMapReduce的分区（Partition）分区发生在<k2,v2>转换为<k3,v3>之间。也就是map输出转换为reduce输入之间。分区是为了能让输入数据能按照一定规则分配给不同的Reducer运行分区数量=Reducer数量MapReduce的分区（Partition）默认情况下根据分区号码=k2.hashCode()%Reducer个数Reducer个数=4分区号码=哈希值%4要求：尽量让数据均匀分布0123481592637哈希值123456789取余123012301MapReduce的分区（Partition）Partitioner的代码实现，需要继承org.apache.hadoop.mapreduce.PartitionerpublicclassMyEmployeeParitionerextendsPartitioner<IntWritable,Employee>{

//numPartition参数：建立多少个分区

@OverridepublicintgetPartition(IntWritablek2,Employeev2,intnumPartition){

//如何建立分区

if(v2.getDeptno()==10){

//放入1号分区中

return1%numPartition;}elseif(v2.getDeptno()==20){

//放入2号分区中

return2%numPartition;}else{

//放入0号分区中

return3%numPartition;}}}MapReduce的分区（Partition）main方法Job设置partitioner和分区数量（分区数量和Reduce任务数量相等）publicclassPartEmployeeMain{publicstaticvoidmain(String[]args)throwsException{

//创建一个jobJobjob=Job.getInstance(newConfiguration());

//指定任务的分区规则

job.setPartitionerClass(MyEmployeeParitioner.class);

//指定建立几个分区

job.setNumReduceTasks(3); ...}MapReduce的合并（Combiner）Combiner是一种特殊Reducer，在Mapper端，先执行一次。它的作用减少Mapper输出到Reduce的数据量，缓解网络传输瓶颈，提高Reducer的执行效率。需要注意的问题：一定要谨慎使用Combiner，有些情况不能使用Combiner，如：求平均值保证引入Combiner以后，不能改变原来的逻辑实例1.WordCount——ShuffleMap阶段分区（partiton）Reduce阶段<k2,v2>k2是单词，v2是单词在每行出现次数合并（combine）——如果有就执行排序与溢写（sortandspill）<k3,v3><“the”,[1]><“there”,[1,1]><“way”,[1]><“where”,[1]><“will”,[1]>按照分区获取溢写出的文件，生成新的键值对Shuffle<“a”,1><“half”,1><“battle”,1><“the”,1><“beginning”,1><“is”,1><“good”,1><“a”,1><“there”,1><“a”,1><“way”,1><“is”,1><“where”,1><“is”,1><“will”,1><“there”,1><“a”,[1,1,1]><“half”,[1]><“battle”,[1]><“beginning”,[1]><“is”,[1,1,1]><“good”,[1]>实例1.WordCount——ShuffleMap阶段分区（partiton）Reduce阶段<k2,v2>k2是单词，v2是单词在每行出现次数合并（combine）排序与溢写（sortandspill）<k3,v3><“the”,[1]><“there”,[2]><“way”,[1]><“where”,[1]><“will”,[1]>按照分区获取溢写出的文件，生成新的键值对Shuffle<“a”,1><“half”,1><“battle”,1><“the”,1><“beginning”,1><“is”,1><“good”,1><“a”,1><“there”,1><“a”,1><“way”,1><“is”,1><“where”,1><“is”,1><“will”,1><“there”,1><“a”,[3]><“half”,[1]><“battle”,[1]><“beginning”,[1]><“is”,[3]><“good”,[1]>MapReduce的合并（Combiner）求平均值的情况下，不能使用CombinerMapper输出123Reducer(1,2,3)2Mapper输出123(1.5,3)不是21.5求平均值求平均值第五章MapReduce5.4

MapReduce工作机制MapReduce作业（Job）运行机制整个过程涉及五个独立的实体：（1）客户端：提交MapReduce作业（2）YARN——ResourceManager：负责协调集群上计算机资源的分配（3）YARN——NodeManager：负责启动和监视集群中机器上的计算容器(container)（4）MapReduce的ApplicationMaster（MRAppMaster），负责协调运行MapReduce作业的任务。它和MapReduce任务在容器中运行，这些容器由资源管理器分配并由节点管理器进行管理。（5）分布式文件系统（一般为HDFS）：共享作业文件MapReduce作业运行机制MapReduce作业运行机制（1）Client请求执行Job。即调用Job的waitForCompletion()，包括：提交Job，并轮询获取、打印进度。（2）向ResourceManager请求获取MapReduce的JobID。（3）计算输入分片。将运行作业所需的资源（Jar文件、配置文件、计算所得分片）保存到HDFS下一个以JobID命名的目录下。（4）调用ResourceManager的submitApplication()提交作业，同时传入（3）的资源。（5）ResourceManager调度器分配一个容器，ResourceManager在NodeManager的管理下，在容器中启动MRAppMaster（6）MRAppMaster对作业初始化MapReduce作业运行机制（接上）（7）MRAppMaster获取HDFS中输入分片,对每个分片创建一个mapTask，和多个reduceTask对象。各个TaskID在此时分配。（8）MRAppMaster决定如何运行作业的各个任务，如果作业很小，就选择同一个JVM上运行。否则，向ResourceManager请求新的容器。（9）当ResourceManager分配了容器，MRAppMaster通过节点管理器启动容器。（10）从HDFS获取作业配置，Jar文件，分片文件，并资源本地化。（11）运行map任务和reduce任务MapReduce进度和状态更新（1）Map任务或Reduce任务运行时，向自己的MRAppMaster报告进度和状态。（2）MRAppMaster形成一个作业的汇聚视图（3）客户端每秒钟轮询一次MRAppMaster获取最新状态。MapReduce进度和状态更新MapReduceprogramJob1a.轮询获取状态clientJVMclientnodeMRAppMasternodemanagernodeNodeManagernodemanagernodeNodeManagerMapTaskorReduceTaskYarnChild3:获取状态1b.向自己的MRAppMaster报告进度和状态2.汇聚作业视图ShuffleMapReduce确保每个Reducer的输入都是按键排序的。系统执行排序、将Mapper输出作为输入传给Reducer的过程称为Shuffle。从许多方面来看，Shuffle是MapReduce的最核心的部分，也被称为“奇迹发生的地方”Shuffle图示——Sort和Spill（溢写）溢写会把内存缓存的数据按分区排序写入文件系统溢写Reduce阶段会把溢写的文件复制并合成一个按分区排序的文件第五章MapReduce5.5

MapReduce编程案例实例1.WordCount——ShuffleMap阶段分区（partiton）Reduce阶段<k2,v2>k2是单词，v2是单词在每行出现次数合并（combine）——如果有就执行排序与溢写（sortandspill）<k3,v3><“the”,[1]><“there”