大数据技术 - 通俗理解MapReduce之WordCount(三)

上一章我们编写了简单的 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(1);
    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(1); // 计数
            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 方法结束任务。

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

 

人已赞赏
站长资讯

BATJ面试必会之 Spring 篇(一)

2020-11-9 3:43:21

站长资讯

Python学习案例之Web版语音合成播报

2020-11-9 3:43:28

0 条回复 A文章作者 M管理员
    暂无讨论,说说你的看法吧
个人中心
购物车
优惠劵
今日签到
有新私信 私信列表
搜索