上一章我们编写了简单的 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 方法结束任务。
以上便是本章的内容，如有疑问、错误、期待的内容，欢迎留言交流。