MR实战：词频统计

文章目录

• 一、实战概述
• 二、提出任务
• 三、完成任务
• • （一）准备数据
  • • 1、在虚拟机上创建文本文件
    • 2、上传文件到HDFS指定目录
  • （二）实现步骤
  • • 1、创建Maven项目
    • 2、添加相关依赖
    • 3、创建日志属性文件
    • 4、创建词频统计映射器类
    • 5、创建词频统计归并器类
    • 6、创建词频统计驱动器类
    • 7、启动应用，查看结果
• 四、实战总结

一、实战概述

• 本实战演练旨在利用Hadoop MapReduce框架在虚拟环境中执行一个简单的词频统计任务。首先，在master节点上创建了一个包含多个单词行的文本文件words.txt，并将该文件上传至HDFS中的指定目录/wordcount/input。

• 在集成开发环境IntelliJ IDEA中，我们创建了一个名为MRWordCount的Maven项目，并引入了Apache Hadoop 3.3.4版本的客户端依赖和JUnit测试框架。为了便于日志管理，添加了log4j.properties配置文件来定义日志输出格式及位置。

• 接下来，实现了两个关键类：WordCountMapper和WordCountReducer。WordCountMapper继承自Mapper接口，负责读取输入文本文件中的每一行内容，将每行按空格分割成单词，并为每个单词生成一个键值对（），以便后续计数处理。而WordCountReducer则继承自Reducer接口，它接收Mapper阶段产生的所有相同单词的键及其对应的次数，进行合并统计并输出格式化的”(单词, 出现次数)”键值对。

• 最后，通过WordCountDriver驱动类完成整个MapReduce作业的设置与执行。此类初始化Hadoop Configuration对象、设置Job参数（包括Mapper和Reducer类、键值类型等）、指定了HDFS上的输入输出路径，并最终提交作业至集群执行。作业完成后，WordCountDriver还会从HDFS上读取结果并显示到控制台。

• 经过上述步骤，当运行WordCountDriver主类时，程序将会读取HDFS上的输入文件，运用MapReduce模型进行分布式计算，最终得到期望的词频统计结果，并在控制台展示出来。这个实例展示了如何使用Hadoop MapReduce进行大规模数据处理的实际操作流程。

二、提出任务

• 单词文件 – words.txt

hello hadoop worldhello hive worldhello hbase worldhadoop hive hbaseI love hadoop and hive

• 使用MR框架，进行词频统计，输出如下结果
  

三、完成任务

（一）准备数据

1、在虚拟机上创建文本文件

• 在master虚拟机上创建words.txt文件
  

2、上传文件到HDFS指定目录

• 创建/wordcount/input目录，执行命令：hdfs dfs -mkdir -p /wordcount/input
  

• 将文本文件words.txt，上传到HDFS的/wordcount/input目录
  

（二）实现步骤

• 说明：集成开发环境IntelliJ IDEA版本 – 2022.3

1、创建Maven项目

• Maven项目 – MRWordCount，设置了JDK版本 – 1.8，组标识 – net.huawei.mr
  

• 单击【Create】按钮，得到初始化项目
  

2、添加相关依赖

• 在pom.xml文件里添加hadoop-client和junit依赖
  

<dependencies>  <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>3.3.4</version> </dependency><dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.13.2</version></dependency></dependencies>

• 刷新项目依赖
  

3、创建日志属性文件

• 在resources目录里创建log4j.properties文件
  

log4j.rootLogger=ERROR, stdout, logfilelog4j.appender.stdout=org.apache.log4j.ConsoleAppenderlog4j.appender.stdout.layout=org.apache.log4j.PatternLayoutlog4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%nlog4j.appender.logfile=org.apache.log4j.FileAppenderlog4j.appender.logfile.File=target/wordcount.loglog4j.appender.logfile.layout=org.apache.log4j.PatternLayoutlog4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

4、创建词频统计映射器类

• 创建net.huawei.mr包，在包里创建WordCountMapper类
  

package net.huawei.mr;import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.mapreduce.Mapper;import org.apache.hadoop.io.Text;import java.io.IOException;/** * 功能：词频统计映射器类 * 作者：华卫 * 日期：2024年01月05日 */public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {@Overrideprotected void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException {// 获取行内容String line = value.toString();// 按空格拆分成单词数组String[] words = line.split(" ");// 遍历单词数组，生成输出键值对for (String word : words) {context.write(new Text(word), new IntWritable(1));}}}

• 该代码定义了一个Hadoop MapReduce作业中的词频统计Mapper类（WordCountMapper），继承自Mapper。它接收输入键值对（LongWritable行偏移量，Text行内容），按空格分割每行文本为单词，并为每个单词输出一个键值对到上下文（），用于后续Reducer进行计数汇总。

• 映射任务与归并任务示意图
  

5、创建词频统计归并器类

• 在net.huawei.mr包里创建WordCountReducer
  

package net.huawei.mr;import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.NullWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;/** * 功能：词频统计归并类 * 作者：华卫 * 日期：2024年01月05日 */public class WordCountReducer extends Reducer<Text, IntWritable, Text, NullWritable> {@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context)throws IOException, InterruptedException {// 定义键（单词）出现次数int count = 0;// 遍历输入值迭代器for (IntWritable value : values) {count = count + value.get(); // 针对此案例，可以写为count++;}// 生成新的键，格式为(word,count)String newKey = "(" + key.toString() + "," + count + ")";// 输出新的键值对context.write(new Text(newKey), NullWritable.get());}}

• 该WordCountReducer类是Hadoop MapReduce中用于词频统计的归约器，继承自Reducer。在reduce方法中，它接收一个单词键（Text类型）及其对应的出现次数迭代器（IntWritable类型）。通过遍历所有次数并将它们累加到变量count上，然后将单词与统计结果拼接成”(word,count)”格式的新键，并使用NullWritable作为值输出，从而实现单词及其词频的合并统计。

6、创建词频统计驱动器类

• 在net.huawei.mr包里，创建WordCountDriver类
  

package net.huawei.mr;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.FSDataInputStream;import org.apache.hadoop.fs.FileStatus;import org.apache.hadoop.fs.FileSystem;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.IOUtils;import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.NullWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Job;import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import java.net.URI;/** * 功能：词频统计驱动器类 * 作者：华卫 * 日期：2024年01月05日 */public class WordCountDriver {public static void main(String[] args) throws Exception {// 创建配置对象Configuration conf = new Configuration();// 设置客户端使用数据节点主机名属性conf.set("dfs.client.use.datanode.hostname", "true");// 获取作业实例Job job = Job.getInstance(conf);// 设置作业启动类job.setJarByClass(WordCountDriver.class);// 设置Mapper类job.setMapperClass(WordCountMapper.class);// 设置map任务输出键类型job.setMapOutputKeyClass(Text.class);// 设置map任务输出值类型job.setMapOutputValueClass(IntWritable.class);// 设置Reducer类job.setReducerClass(WordCountReducer.class);// 设置reduce任务输出键类型job.setOutputKeyClass(Text.class);// 设置reduce任务输出值类型job.setOutputValueClass(NullWritable.class);// 定义uri字符串String uri = "hdfs://master:9000";// 创建输入目录Path inputPath = new Path(uri + "/wordcount/input");// 创建输出目录Path outputPath = new Path(uri + "/wordcount/output");// 获取文件系统FileSystem fs = FileSystem.get(new URI(uri), conf);// 删除输出目录（第二个参数设置是否递归）fs.delete(outputPath, true);// 给作业添加输入目录（允许多个）FileInputFormat.addInputPath(job, inputPath);// 给作业设置输出目录（只能一个）FileOutputFormat.setOutputPath(job, outputPath);// 等待作业完成job.waitForCompletion(true);// 输出统计结果System.out.println("======统计结果======");FileStatus[] fileStatuses = fs.listStatus(outputPath);for (int i = 1; i < fileStatuses.length; i++) {// 输出结果文件路径System.out.println(fileStatuses[i].getPath());// 获取文件系统数据字节输入流FSDataInputStream in = fs.open(fileStatuses[i].getPath());// 将结果文件显示在控制台IOUtils.copyBytes(in, System.out, 4096, false);}}}

• 该WordCountDriver类是Hadoop MapReduce框架中用于执行词频统计任务的驱动类。

1. 初始化配置：首先创建一个Configuration对象，设置客户端使用数据节点主机名属性以便正确解析路径。

2. 构建作业实例：通过Job.getInstance(conf)获取一个MapReduce作业实例，并将当前类（WordCountDriver）作为作业启动类，这样Hadoop在运行时能识别到主程序入口。

3. 配置Mapper和Reducer：分别指定Map阶段使用的类为WordCountMapper，Reduce阶段使用的类为WordCountReducer。同时设定Map阶段输出键值对类型为Text和IntWritable，Reduce阶段输出键值对类型也为Text和NullWritable。

4. 定义文件系统URI：设置HDFS地址为hdfs://master:9000，并基于此URI创建输入目录（/wordcount/input）和输出目录（/wordcount/output）。

5. 文件系统操作：连接到HDFS文件系统，删除已存在的输出目录以准备新的计算结果，然后向作业添加输入目录和设置输出目录。

6. 提交并监控作业：调用job.waitForCompletion(true)方法提交作业并等待其完成。当作业完成后，会返回一个布尔值表示作业是否成功执行。

7. 读取并显示结果：作业结束后，列出输出目录下的所有文件，遍历这些文件并打开每个文件进行读取。利用IOUtils.copyBytes方法将结果文件的内容复制到控制台输出，展示词频统计的结果。

• 总之，WordCountDriver类负责整个词频统计任务的初始化、配置、执行以及结果展示工作，它将Hadoop MapReduce的各个组件如Mapper、Reducer与实际的输入输出路径关联起来，形成了一个完整的词频统计应用。

7、启动应用，查看结果

• 运行WordCountDriver类，查看结果
  

四、实战总结

• 本实战通过Hadoop MapReduce框架，在虚拟环境中对words.txt文件进行了词频统计。首先，将数据上传至HDFS，并在IntelliJ IDEA中创建Maven项目配置相关依赖。实现的WordCountMapper负责按空格拆分单词并初始化词频为1，WordCountReducer则对相同单词的计数进行合并。最后，WordCountDriver类配置作业参数、指定输入输出路径并在集群上执行任务，完成后从HDFS读取并展示统计结果。整个过程演示了MapReduce模式处理文本数据进行词频统计的完整流程。