MapReduce WordCount案例分析以及代码实现

青旅半醒 2023-07-13 15:01 7阅读 0赞

环境参数：hadoop3.2.1、IDEA、maven3.6

### **案例分析** ###

WordCount程序是学习Hadoop的入门程序，我们有必要详解一下。能够完整的运行WordCount程序需要如下步骤：本地的文本文件上传到HDFS上，WordCount程序实现MapReduce过程，输出结果到HDFS上。

首先来看一下WordCount简化的MapReduce工作过程(这个过程简化了许多)：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p6cWFhYXNzcw_size_16_color_FFFFFF_t_70][]

1、将数据文件拆分成splits，由于测试用的文件较小，所以每个文件为一个split，并将文件按行分割形成<key, value>，这一步由MapReduce框架自动完成，其中偏移量(即k1值)包括了回车所占的字符数(Windows和Linux环境会不同)。

2、将分割好的<key, value>对交给用户定义的map方法进行处理，生成新的<k2, v2>对

3、得到map方法输出的<k2, v2>之后，Mapper会将它们按照key值进行排序，并执行Combine过程，将key值相同的value值累加，得到Mapper的最终输出结果<k2,\{v2..\}>

4、Reducer先对Mapper接收的数据进行排序（这一步上图省略），再交由用户自定义的reduce方法进行处理，得到新的<k3, v3>对，并作为wordcount的输出结果

### **代码实现** ###

最简单的MapReduce应用程序至少包含 3 个部分：一个 Map 函数、一个 Reduce 函数和一个 main 函数。在运行一个mapreduce计算任务时候，任务过程被分为两个阶段：map阶段和reduce阶段，每个阶段都是用键值对（key/value）作为输入（input）和输出（output）。main 函数将作业控制和文件输入/输出结合起来。

首先我们准备一份单词数据文件并上传到hdfs文件系统中。

**1、文件内容**

word.txt

hello world
    hello hadoop
    the world is beautiful

**2、上传文件到hdfs**

hdfs dfs -put word.txt /

**3、新建一个maven项目并导入hadoop-client依赖**

<dependencies>
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-client</artifactId>
                <version>3.2.1</version>
                <scope>provided</scope>
            </dependency>
        </dependencies>

注意：上面导入的版本要和你安装的hadoop版本一致，还有把<scop>值设置为provided是为了在导出jar包的时候不把hadoop-client加进去，以免增加jar大小。

**4、java代码**

import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.Mapper;
    import org.apache.hadoop.mapreduce.Reducer;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    import java.io.IOException;
    
    /**
     * 单词计数
     * 需求：读取hdfs上的word.txt文件，计算出文件中单词出现的总次数
     */
    public class WordCountJob{
        /**
         * 创建自定义的Mapper类
         * 四个参数分别是（map读入的偏移量类型，值类型，map即将输出的键类型值类型）
         */
        public static class MyMapper extends Mapper<LongWritable,Text,Text,LongWritable>{
            /**
             * map阶段的业务逻辑就写在自定义的map()方法中
             * 这个map函数就是接受<k1,v1>产生<k2,v2>
             * @param k1 每一行的行首偏移量
             * @param v1 每一行内容
             * @param context 收集Mapper输出的<k,v>对
             * @throws IOException
             * @throws InterruptedException
             */
            @Override
            protected void map(LongWritable k1, Text v1, Context context) throws IOException, InterruptedException {
                //把读入的一行数据按空格切割
                String[] words = v1.toString().split(" ");
                //迭代切割出来的单词数据
                for (String word:words){
                    //把迭代出来的单词封装成<k2,v2>
                    Text k2 = new Text(word);
                    LongWritable v2 = new LongWritable(1L);
                    //把<k2,v3>写出去
                    context.write(k2,v2);
                }
            }
        }
    
        /**
         * 创建自定义的MyReduce类
         * 四个参数分别是（map输出的键类型，值类型，reducer即将输出的键类型值类型）
         */
        public static class MyReduce extends Reducer<Text,LongWritable,Text,LongWritable>{
            /**
             * 针对v2s的数据累加求和并且把数据封装成<k3,v3>写出去
             *
             * @param k2 每一个单词
             * @param v2s 相同单词的value列表
             * @param context 收集reduce输出的<k,v>对
             * @throws IOException
             * @throws InterruptedException
             */
            @Override
            protected void reduce(Text k2, Iterable<LongWritable> v2s, Context context) throws IOException, InterruptedException {
                //创建一个sum变量，保存v2s的和
                Long sum =0L;
                //迭代出相同key的value并求和
                for (LongWritable v2:v2s){
                    sum += v2.get();
                }
                LongWritable v3 = new LongWritable(sum);
                //把<k3,v3>写出去，这里的看k2其实就已经是v3了
                context.write(k2,v3);
            }
        }
    
        /**
         * 组装job
         * @param args [0]输入路径 [1]输出路径
         */
        public static void main(String[] args) {
            //job需要的配置参数
            Configuration conf = new Configuration();
            try {
                //这里是为了防止没有写入输入和输出路径
                if (args.length!=2){
                    System.exit(100);
                }
                //创建一个job
                Job job = Job.getInstance(conf);
                //注意：这一行必须设置，否则在集群中找不到WordCountJob类
                job.setJarByClass(WordCountJob.class);
                //指定输入路径（可以是文件，也可以是文件）路径参数从启动任务的时候传进来
                FileInputFormat.setInputPaths(job,new Path(args[0]));
                //指定输出文件路径（只能指定一个不存在的目录）
                FileOutputFormat.setOutputPath(job,new Path(args[1]));
                //指定map所在类
                job.setMapperClass(MyMapper.class);
                //指定k2的类型
                job.setMapOutputKeyClass(Text.class);
                //指定v2的类型
                job.setMapOutputValueClass(LongWritable.class);
                //指定reduce所在类
                job.setReducerClass(MyReduce.class);
                //指定k3类型
                job.setOutputKeyClass(Text.class);
                //指定v3类型
                job.setOutputValueClass(LongWritable.class);
                //提交作业
                job.waitForCompletion(true);
            } catch (IOException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ClassNotFoundException e) {
                e.printStackTrace();
            }
        }
    }

**5、导出jar包**

在idea中直接使用maven的package即可导出来

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p6cWFhYXNzcw_size_16_color_FFFFFF_t_70 1][]

导出成功后再当前项目的target目录下会产生一个jar包。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p6cWFhYXNzcw_size_16_color_FFFFFF_t_70 2][]

然后上传到hadoop主节点所在的服务器即可。

**6、上传jar包**

可以利用ftp工具这里上传到hadoop主节点所在的服务器

**7、提交任务**

hadoop jar hadoopdemo-1.0-SNAPSHOT.jar com.hadoop.demo.WordCountJob hdfs://hadoop1:9000/word.txt hdfs://hadoop1:9000/out

![20200312234746872.PNG][]

指令解释：

hadoop jar :使用hadoop运行jar包

hadoopdemo-1.0-SNAPSHOT.jar ：之前我们到出的项目jar包

com.hadoop.demo.WordCountJob ：主入口类所在的类全名（加上类所在的包名，如果没有包写类名即可）

hdfs://hadoop1:9000/word.txt ：输入文件

hdfs://hadoop1:9000/out ：输出文件到该目录，注意：此目录一定是不存在的目录

运行输出如下即job执行完毕：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p6cWFhYXNzcw_size_16_color_FFFFFF_t_70 3][]

** 8、查看输出结果**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p6cWFhYXNzcw_size_16_color_FFFFFF_t_70 4][]

到此一个基本的MapReduce程序就完成了！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p6cWFhYXNzcw_size_16_color_FFFFFF_t_70]: /images/20230529/2a31548c9c00423ab0ecda2151f848f2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p6cWFhYXNzcw_size_16_color_FFFFFF_t_70 1]: /images/20230529/7cfe273c88e649ea8bcde53babab7509.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p6cWFhYXNzcw_size_16_color_FFFFFF_t_70 2]: /images/20230529/8d2d86e61f89476abc1f4b89fcac4b8e.png
[20200312234746872.PNG]: /images/20230529/af24a15185c34059b587af0ae307ac4f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p6cWFhYXNzcw_size_16_color_FFFFFF_t_70 3]: /images/20230529/a5b65cd035f24f94b023fa8db39c3b88.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p6cWFhYXNzcw_size_16_color_FFFFFF_t_70 4]: /images/20230529/e111a24e6276455c8ae721deeb367a8d.png