MR实战:词频统计

2024-01-09 19:16:58

一、实战概述

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

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

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

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

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

二、提出任务

  • 单词文件 - words.txt
hello hadoop world
hello hive world
hello hbase world
hadoop hive hbase
I 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-clientjunit依赖
    在这里插入图片描述
<dependencies>                                   
    <!--hadoop客户端-->                             
    <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, logfile
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/wordcount.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.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> {
    @Override
    protected 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, Text, IntWritable>。它接收输入键值对(LongWritable行偏移量,Text行内容),按空格分割每行文本为单词,并为每个单词输出一个键值对到上下文(<Text: 单词, IntWritable: 1>),用于后续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> {
    @Override
    protected 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<Text, IntWritable, Text, NullWritable>。在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阶段输出键值对类型为TextIntWritable,Reduce阶段输出键值对类型也为TextNullWritable

  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模式处理文本数据进行词频统计的完整流程。

文章来源:https://blog.csdn.net/howard2005/article/details/135415704
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。