大数据技术 - 通俗理解MapReduce之WordCount（三）

时间：2023-11-23 19:41:08

上一章我们编写了简单的 MapReduce 程序，掌握这些就能编写大多数数据处理的代码。但是 MapReduce 框架提供给用户的能力并不止如此，本章我们仍然以上一章 word count 为例，继续完善我们的数据处理代码。本章主要关注的重点包括三个部分：

　　1. 完整的 map / reduce 任务，完整的 map 任务除了 map 方法里的逻辑外，还包括任务运行前的准备工作以及任务结束后的清理工作，reduce 任务也一样

　　2. Counter 的作用，有时候为了统计程序运行中任务的状态，比如：某个异常出现的次数，因此需要一个计数器进行统计并输出

　　3. 给 MapReduce 任务传自定义配置，命令行可以实现传参数，但是参数比较多的情况下，命令行参数不好维护且不具备很好的可读性，最好能够使用 Hadoop 配置文件中的那种格式配置

在这里， 我们仍然用上一章 word count 中的 map 任务， 区别是我们可以通过自定义配置实现只统计某个单词出现的次数，同时增加了计数功能。下面看下如何在 map 任务中实现上面这三个内容。

package com.cnblogs.duma.mapreduce;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Counter;

import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**

 * WordCountMapper 继承 Mapper 类，需要指定4个泛型类型，分别是

 * 输入 key 类型：本例中输入的 key 为每行文本的行号，例子中用不到所以这里是 Object

 * 输入 value 类型：本例中输入的 value 是每行文本，因此是Text

 * 输出 key 类型：map 输出的是每个单词，类型为 Text

 * 输出 value 类型：单词出现的次数，为 1，因此类型 IntWritable

 */

public class WordCountMapper

        extends Mapper<Object, Text, Text, IntWritable> {

    /**

     * 把每个单词映射成 <word, 1> 的格式

     */

    private final static IntWritable one = new IntWritable();

    private Text outWord = new Text();

    private String filterWord;

    /**

     * 每一个 map 进程调用一次

     * @param context

     * @throws IOException

     * @throws InterruptedException

     */

    @Override

    protected void setup(Context context) throws IOException, InterruptedException {

        Configuration conf = context.getConfiguration();

        filterWord = conf.get("wordcount.filter.word", null);

        /**

         * 初始化工作， 比如连接数据库

         */

    }

    /**

     * 每个 map 方法处理一行数据

     * @param key    输入的行号

     * @param value  每一行文本

     * @param context

     * @throws IOException

     * @throws InterruptedException

     */

    @Override

    protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {

        String[] words = value.toString().split(" "); //空格分割一行中的每个单词

        Counter counter = context.getCounter("group1", "counter1"); //第一个参数代表计数组，第二个参数代表计数名称

        for (String word : words) {

            if (filterWord != null && !filterWord.equals(word)) //判断是否只统计过滤词

                continue;

            counter.increment(); // 计数

            outWord.set(word);

            context.write(outWord, one); // map输出

        }

    }

    @Override

    protected void cleanup(Context context) throws IOException, InterruptedException {

        /**

         * 做清理工作， 比如释放数据库连接

         */

    }

}

WordCounterMapper类的定义与上一章一样，只是针对上述三个问题做出了相应的修改：

　　1. 增加 setup 和 cleanup 方法，假设 map 任务中需要获取数据库里的内容，由于连接数据库的操作需要消耗资源，且每个 map 任务连接一次数据库即可，因此连接数据库的操作可以写在 setup 方法中。setup 方法在 map 方法开始前调用一次。同理 cleanup 方法在 map 方法执行完后调用一次，释放数据库连接防止内存泄漏。

　　2. 在 map 方法中增加了计数的功能，counter.increment(1);，该计数器为了统计 map 方法中具体处理了多少个单词。实际应用中我们可能想对抛异常的业务逻辑做计数，这样日后可以通过分析 MapReduce 输出的计数日志，来统计出错的次数。

　　3. setup 方法中有个获取配置的代码，wordcount.filter.word 代表只对某个词做统计，这个配置是我们自定义的，看上去跟 Hadoop 系统配置写法类似，这种写法见名知意，方便维护。

为了上述代码顺利执行，我们需要修改下驱动程序

package com.cnblogs.duma.mapreduce;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.IntWritable;

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 org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;

public class WordCount {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        Configuration conf = new Configuration();

        GenericOptionsParser optionParser = new GenericOptionsParser(conf, args); //识别命令行参数中的自定义配置

        String[] remainingArgs = optionParser.getRemainingArgs(); //获取处理自定义配置外的其他参数

        ...

        FileInputFormat.addInputPath(job, new Path(remainingArgs[0])); //增加输入文件

        FileOutputFormat.setOutputPath(job, new Path(remainingArgs[1])); //设置输出目录

        System.exit(job.waitForCompletion(true) ? 0 : 1);

    }

代码跟上一章基本一致， 增加了能够处理自定义配置的代码。打包，上传到 hadoop0 机器， 执行以下命令启动程序

hadoop jar hadoop-ex-1.0-SNAPSHOT.jar com.cnblogs.duma.mapreduce.WordCount -Dwordcount.filter.word=hadoop /hadoop-ex/wordcount/input /hadoop-ex/wordcount/output

首先，我们看到任务的启动命令比上一章多了 -Dwordcount.filter.word=hadoop，-D 后面跟的就是配置的 key，= 后面跟的是配置的 value。这个配置让我们的程序只统计 hadoop 这个词出现的次数。如果驱动程序中不加 new GenericOptionsParser(conf, args) 这段代码，那么 -Dwordcount.filter.word=hadoop 就被看做是普通命令行参数， args[0] = "-Dwordcount.filter.word=hadoop" 。有兴趣的读者可以自己试试，同时也可以看下 GenericOptionsParser 的代码。任务执行完可以看到以下的输出信息

Map-Reduce Framework

        Map input records=4

        Map output records=3

        Map output bytes=33

        Map output materialized bytes=51

        Input split bytes=233

        Combine input records=0

        Combine output records=0

        Reduce input groups=1

        Reduce shuffle bytes=51

        Reduce input records=3

        Reduce output records=1

        Spilled Records=6

        Shuffled Maps =2

        Failed Shuffles=0

        Merged Map outputs=2

        GC time elapsed (ms)=981

        CPU time spent (ms)=2390

        Physical memory (bytes) snapshot=540434432

        Virtual memory (bytes) snapshot=6240616448

        Total committed heap usage (bytes)=262197248

    Shuffle Errors

        BAD_ID=0

        CONNECTION=0

        IO_ERROR=0

        WRONG_LENGTH=0

        WRONG_MAP=0

        WRONG_REDUCE=0

    group1

        counter1=3

可以看到 MapReduce 有系统默认的计数器， 比如：Map input records=4 代表一共输入了 4 条记录。我们能看到自定义的计数器 counter1=3 ，之前解释过这个代表一共处理了多少个词， 这里是 3 也是对的， 因为 hadoop 这个词出现了 3 次。再看下输出结果

[root@hadoop0 hadoop-ex]# hadoop fs -cat /hadoop-ex/wordcount/output/*

hadoop    3

输出是符合预期的，只统计了 hadoop 这个词且出现了 3 次。

以上便是这章的主要内容，这里虽然以 map 任务举例，但这三点同样可以在 reduce 任务中应用。掌握这些内容编写大多数 MapReduce 任务基本没有问题。

番外

这篇文章我们知道一个 map 或 reduce 任务只会执行一次 setup 和 cleanup 方法，我们实际看下 Hadoop 源码中是如何实现的。以 WordCountMapper 为例， 它继承了 Mapper 类， 在 Mapper 类中有一个 run 方法

/**

   * Expert users can override this method for more complete control over the

   * execution of the Mapper.

   * @param context

   * @throws IOException

   */

  public void run(Context context) throws IOException, InterruptedException {

    setup(context);

    try {

      while (context.nextKeyValue()) {

        map(context.getCurrentKey(), context.getCurrentValue(), context);

      }

    } finally {

      cleanup(context);

    }

  }

可以看到， 函数第一行调用一次 setup 方法，循环获取每个 key - value 对，即 <行号， 每行文本> 调用 map 方法进行处理。处理完毕后调用 cleanup 方法结束任务。

以上便是本章的内容，如有疑问、错误、期待的内容，欢迎留言交流。

相关文章