单元5 任务5.2-MapReduce实现词频统计

《大数据平台部署与运维》单元5MapReduce实现电商销售数据统计任务5.2MapReduce实现词频统计01理解MapReduce数据处理流程02掌握MapReduce相关Java

API学习目标03掌握MapReduce驱动类的编写任务5.2MapReduce实现词频统计【任务场景】经理：小张，我们后续也要使用Java编写MapReduce程序，你对MapReduce的JavaAPI了解吗？小张：Hadoop框架底层就是用Java实现的，所以Hadoop提供了许多用来调用MapReduce的JavaAPI，我们可以借助MapReduce的JavaAPI可以完成很多项自定义操作。经理：是的，使用MapReduce的JavaAPI时也要注意数据类型转换，要使用Hadoop特有的Writable数据类型，你先熟悉一下MapReduce的JavaAPI，然后自己实现一个词频统计程序，在服务器上运行一下试试吧。小张：好的。任务5.2MapReduce实现词频统计【任务布置】MapReduce的运行依赖与JDK和Hadoop，因此必须将Hadoop的基础环境提前安装好，才能进行MapReduce的运行和操作。本任务要求在前面已经完成安装部署Hadoop平台的node1节点上完成，要求理解MapReduce数据处理流程，掌握MapReduce相关JavaAPI及驱动类的创建；基于IDEA进行开发，最终在node1上运行打包后的程序，实现词频统计。任务5.2MapReduce实现词频统计5.2.1MapReduce数据处理流程MapReduce处理数据过程主要分成Map和Reduce两个阶段。数据处理流程如下图所示：任务5.2MapReduce实现词频统计Map任务处理（共五步）（1）读取输入文件内容，解析成key、value对。对输入文件的每一行，解析成key、value对。每一个键值对调用一次map函数。（2）写自己的逻辑，对输入的key、value处理，转换成新的key、value输出。（3）对输出的key、value进行分区。（4）对不同分区的数据，按照key进行排序、分组。相同key的value放到一个集合中。（5）(可选)分组后的数据进行归约。任务5.2MapReduce实现词频统计Reduce任务处理（共三步）（1）对多个map任务的输出，按照不同的分区，通过网络copy到不同的reduce节点。（2）对多个map任务的输出进行合并、排序。写reduce函数自己的逻辑，对输入的key、value处理，转换成新的key、value输出。（3）把reduce的输出保存到文件中。任务5.2MapReduce实现词频统计5.2.2MapReduce相关Java

API及应用1.

MapReduce中数据类型介绍MapReduce中所有的数据类型都要实现Writable接口，以便于这些类型定义的数据可以被序列化进行网络传输和文件存储。Hadoop数据类型都放在org.apache.hadoop.io包下。Hadoop数据类型和Java数据类型对照表数据类型Hadoop数据类型Java数据类型布尔型BooleanWritableboolean整型ByteWritablebyteShortWritableshortIntWritableintLongWritablelong浮点型FloatWritablefloatDoubleWritabledouble字符串（文本）Textstring数组ArrayWritableArraymap集合MapWritablemap任务5.2MapReduce实现词频统计2.

MapReduce

Mapper类在MapReduce中，Mapper类的作用是将输入的键值对映射到一组中间键值对。它将输入记录转换为中间记录。这些中间记录与给定的输出键相关联，并传递给Reducer以获得最终输出。Mapper类的常用方法如下表所示：方法描述voidcleanup(Contextcontext)此方法仅在任务结束时调用一次。voidmap(KEYINkey,VALUEINvalue,Contextcontext) 对于输入Mapper中的每个键值，只能调用此方法一次。voidrun(Contextcontext)可以重写此方法以控制Mapper的执行。voidsetup(Contextcontext)此方法仅在任务开始时调用一次。任务5.2MapReduce实现词频统计3.

MapReduce

Reducer类在MapReduce中，Reducer类的作用是减少中间值的集合。它的实现可以通过JobContext.getConfiguration()方法访问作业的配置。Reducer类的常用方法如下表所示：方法描述voidcleanup(Contextcontext)此方法仅在任务结束时调用一次。voidreduce(KEYINkey,Iterable<VALUEIN>values,Contextcontext) 此方法只为每个键调用一次。voidrun(Contextcontext)这个方法可以用来控制Reducer的任务。voidsetup(Contextcontext)此方法仅在任务开始时调用一次。任务5.2MapReduce实现词频统计4.

MapReduce

Job类Job类用于配置作业和提交作业，它还控制执行和查询状态。Job类的常用方法如下表所示：方法描述voidsetJarByClass(Class<>class)核心接口，指定执行类所在的jar包本地位置。java通过class文件找到执行jar包，该jar包被上传到HDFS。voidsetMapOutputKeyClass(Class<>class)该方法用于设置map输出数据的key类型。voidsetMapOutputValueClass(Class<>class)该方法用于设置map输出数据的value类型。setOutputKeyClass(Class<>theClass)核心接口，指定MapReduce作业的输出key的类型setOutputValueClass(Class<>theClass)核心接口，指定MapReduce作业的输出value的类型voidsetMapperClass(Class<extendsMapper>class)核心接口，指定MapReduce作业的Mapper类，默认为空。voidsetNumReduceTasks(inttasks)该方法用于设置job的reduce任务数voidsetReducerClass(Class<extendsReducer>class)核心接口，指定MapReduce作业的Reducer类，默认为空。setPartitionerClass(Class<extendsPartitioner>class)指定MapReduce作业的Partitioner类。该方法用来分配map的输出结果到哪个reduce类，默认使用HashPartitioner，均匀分配map的每条键值对记录。任务5.2MapReduce实现词频统计5.2.3MapReduce驱动类驱动类主要用于关联Mapper和Reducer以及提交整个程序。驱动类的开发有着固定的格式。编写驱动类总分七大步骤：1.获取job对象。新建一个配置对象实例，用于获取job实例方法传递参数。2.设置Jar的路径(就是当前驱动类)。设置当前驱动类路径，参数即为当前驱动类的类名。Configurationconfiguration=newConfiguration();Jobjob=Job.getInstance(configuration);job.setJarByClass(WordCountDriver.class);任务5.2MapReduce实现词频统计3.关联mapper和reducer。设置mapper和reducer类，参数为自定义的mapper类类名及reducer类的类名。4.设置mapper输出的key和value类型。设置mapper输出的key和value类型时可根据业务灵活调整，类型需使用已实现Writable的数据类型。5.设置最终输出的key和value设置最终输出的key和value类型时可根据业务灵活调整，需使用已实现Writable的Hadoop数据类型。job.setMapperClass(WordcountMapper.class);job.setReducerClass(WordcountReducer.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);

任务5.2MapReduce实现词频统计6.设置输入输出路径(要读取的数据输入路径和输出数据的路径)设置路径有两种方式，一种是通过传入参数动态读取，另一种是在程序中写成固定路径，但是会降低扩展性，推荐使用第一种方式。第一种方式在运行时必须传入输入输出路径，示例：hadoopjarWordCountApp/hello/out需要注意这里的路径都是HDFS中的路径，也就是读取和输出的文件都是在HDFS中的。7.提交job将作业提交到群集并等待它完成，参数设置为true代表打印显示对应的进度，根据作业结果，终止当前运行的Java虚拟机，退出程序。FileInputFormat.setInputPaths(job,newPath(args[1]));FileOutputFormat.setOutputPath(job,newPath(args[2]));//或FileInputFormat.setInputPaths(job,newPath("/map/input"));FileOutputFormat.setOutputPath(job,newPath("/map/output"));

Booleanresult=job.waitForCompletion(true);System.exit(result?0:-1);

任务5.2MapReduce实现词频统计【工作流程】创建新的项目，编写Mapper类，继承父类并重写map方法，编写Reducer类，继承父类并重写reduce方法，编写驱动类，程序导出为jar包运行，查看结果。任务5.2MapReduce实现词频统计【操作步骤】创建新的项目，项目类型选择Maven，项目名称为MRWordCount，pom依赖为以下内容。<dependencies><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>RELEASE</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-hdfs</artifactId><version>3.1.1</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>3.1.1</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>3.1.1</version></dependency><dependency><groupId>org.apache.logging.log4j</groupId><artifactId>log4j-core</artifactId><version>2.8.2</version></dependency></dependencies>

任务5.2MapReduce实现词频统计1.编写Mapper类创建Java类MyMapper，继承父类Mapper，重写父类map方法，实现如下代码：importorg.apache.hadoop.io.LongWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Mapper;publicclassMyMapperextendsMapper<LongWritable,Text,Text,LongWritable>{ protectedvoidmap(LongWritablek1,Textv1,Contextcontext) throwsjava.io.IOException,InterruptedException{//对v1用空格进行分割，返回单词数组 Stringwords[]=v1.toString().split(""); Textk2=newText(); LongWritablev2=newLongWritable(1); for(Stringword:words) { k2.set(word); context.write(k2,v2);//将k2/v2写入磁盘 } };}

任务5.2MapReduce实现词频统计2.编写Reducer类创建Java类MyReducer，继承父类Reducer，重写父类reduce方法，实现如下代码：importorg.apache.hadoop.io.LongWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Reducer;

publicclassMyReducerextendsReducer<Text,LongWritable,Text,LongWritable>{ protectedvoidreduce(Textk2,Iterable<LongWritable>vs2,Contextcontext) throwsjava.io.IOException,InterruptedException{

Textk3=k2;//k3和k2一样，表示每个单词 longsum=0; for(LongWritablei:vs2) { sum+=i.get(); } LongWritablev3=newLongWritable(sum);

context.write(k3,v3);//最终结果

};}

任务5.2MapReduce实现词频统计3.创建驱动类WordCountApp创建驱动类WordCountApp，并编码如下代码：importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.io.LongWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Job;importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;importorg.apache.hadoop.conf.Configuration;publicclassWordCountApp{ publicstaticvoidmain(String[]args)throwsIOException,ClassNotFoundException,InterruptedException{ Configurationconf=newConfiguration(); Jobjob=Job.getInstance(conf,"myjob"); job.setJarByClass(WordCountApp.class); PathfileIn=newPath(args[1]);

PathfileOut=newPath(args[2]);

job.setMapperClass(MyMapper.class);//设置使用的Mapper类

job.setMapOutputKeyClass(Text.class);//设置k2/v2类型

job.setMapOutputValueClass(LongWritable.class); FileInputFormat.setInputPaths(job,fileIn);//设置任务的输入文件路径

job.setReducerClass(MyReducer.class);//设