官网MapReduce实例代码详细批注

387次阅读

没有评论

共计 7724 个字符，预计需要花费 20 分钟才能阅读完成。

引言

1. 本文不描述 MapReduce 入门知识，这类知识网上很多，请自行查阅

2. 本文的实例代码来自官网

http://hadoop.apache.org/docs/current/hadoop-mapreduce-client/hadoop-mapreduce-client-core/MapReduceTutorial.html

最后的 WordCount v2.0，该代码相比源码中的 org.apache.Hadoop.examples.WordCount 要复杂和完整，更适合作为 MapReduce 模板代码

3. 本文的目的就是为开发 MapReduce 的同学提供一个详细注释了的模板，可以基于该模板做开发。

——————————————————————————–

官网实例代码（略有改动）

WordCount2.java

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.StringTokenizer;
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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.StringUtils;
public class WordCount2 {
// 日志组名 MapCounters，日志名 INPUT_WORDS
static enum MapCounters {
INPUT_WORDS
}
static enum ReduceCounters {
OUTPUT_WORDS
}

// static enum CountersEnum {INPUT_WORDS,OUTPUT_WORDS}
// 日志组名 CountersEnum，日志名 INPUT_WORDS 和 OUTPUT_WORDS

public static class TokenizerMapper extends
Mapper<Object, Text, Text, IntWritable> {
private final static IntWritable one = new IntWritable(1); // map 输出的 value
private Text word = new Text(); // map 输出的 key
private boolean caseSensitive; // 是否大小写敏感，从配置文件中读出赋值
private Set<String> patternsToSkip = new HashSet<String>(); // 用来保存需过滤的关键词，从配置文件中读出赋值
private Configuration conf;
private BufferedReader fis; // 保存文件输入流
/**
* 整个 setup 就做了两件事：1. 读取配置文件中的 wordcount.case.sensitive，赋值给 caseSensitive 变量
* 2. 读取配置文件中的 wordcount.skip.patterns，如果为 true，将 CacheFiles 的文件都加入过滤范围
*/
@Override
public void setup(Context context) throws IOException,
InterruptedException {
conf = context.getConfiguration();
// getBoolean(String name, boolean defaultValue)
// 获取 name 指定属性的值，如果属性没有指定，或者指定的值无效，就用 defaultValue 返回。
// 属性可以在命令行中通过 -Dpropretyname 指定，例如 -Dwordcount.case.sensitive=true
// 属性也可以在 main 函数中通过 job.getConfiguration().setBoolean(“wordcount.case.sensitive”,
// true) 指定
caseSensitive = conf.getBoolean(“wordcount.case.sensitive”, true); // 配置文件中的 wordcount.case.sensitive 功能是否打开
// wordcount.skip.patterns 属性的值取决于命令行参数是否有 -skip，具体逻辑在 main 方法中
if (conf.getBoolean(“wordcount.skip.patterns”, false)) {// 配置文件中的 wordcount.skip.patterns 功能是否打开
URI[] patternsURIs = Job.getInstance(conf).getCacheFiles(); // getCacheFiles() 方法可以取出缓存的本地化文件，本例中在 main 设置
for (URI patternsURI : patternsURIs) {// 每一个 patternsURI 都代表一个文件
Path patternsPath = new Path(patternsURI.getPath());
String patternsFileName = patternsPath.getName().toString();
parseSkipFile(patternsFileName); // 将文件加入过滤范围，具体逻辑参见 parseSkipFile(String
// fileName)
}
}
}
/**
* 将指定文件的内容加入过滤范围
*
* @param fileName
*/
private void parseSkipFile(String fileName) {
try {
fis = new BufferedReader(new FileReader(fileName));
String pattern = null;
while ((pattern = fis.readLine()) != null) {// SkipFile 的每一行都是一个需要过滤的 pattern，例如 \!
patternsToSkip.add(pattern);
}
} catch (IOException ioe) {
System.err
.println(“Caught exception while parsing the cached file ‘”
+ StringUtils.stringifyException(ioe));
}
}
@Override
public void map(Object key, Text value, Context context)
throws IOException, InterruptedException {
// 这里的 caseSensitive 在 setup() 方法中赋值
String line = (caseSensitive) ? value.toString() : value.toString()
.toLowerCase(); // 如果设置了大小写敏感，就保留原样，否则全转换成小写
for (String pattern : patternsToSkip) {// 将数据中所有满足 patternsToSkip 的 pattern 都过滤掉
line = line.replaceAll(pattern, “”);
}
StringTokenizer itr = new StringTokenizer(line); // 将 line 以 \t\n\r\f 为分隔符进行分隔
while (itr.hasMoreTokens()) {
word.set(itr.nextToken());
context.write(word, one);
// getCounter(String groupName, String counterName) 计数器
// 枚举类型的名称即为组的名称，枚举类型的字段就是计数器名称
Counter counter = context.getCounter(
MapCounters.class.getName(),
MapCounters.INPUT_WORDS.toString());
counter.increment(1);
}
}
}
/**
* Reducer 没什么特别的升级特性
*
* @author Administrator
*/
public static class IntSumReducer extends
Reducer<Text, IntWritable, Text, IntWritable> {
private IntWritable result = new IntWritable();
public void reduce(Text key, Iterable<IntWritable> values,
Context context) throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values) {
sum += val.get();
}
result.set(sum);
context.write(key, result);
Counter counter = context.getCounter(
ReduceCounters.class.getName(),
ReduceCounters.OUTPUT_WORDS.toString());
counter.increment(1);
}
}
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
GenericOptionsParser optionParser = new GenericOptionsParser(conf, args);
/**
* 命令行语法是：hadoop command [genericOptions] [application-specific
* arguments] getRemainingArgs() 取到的只是 [application-specific arguments]
* 比如：$ bin/hadoop jar wc.jar WordCount2 -Dwordcount.case.sensitive=true
* /user/joe/wordcount/input /user/joe/wordcount/output -skip
* /user/joe/wordcount/patterns.txt
* getRemainingArgs() 取到的是 /user/joe/wordcount/input
* /user/joe/wordcount/output -skip /user/joe/wordcount/patterns.txt
*/
String[] remainingArgs = optionParser.getRemainingArgs();
// remainingArgs.length == 2 时，包括输入输出路径：
///user/joe/wordcount/input /user/joe/wordcount/output
// remainingArgs.length == 4 时，包括输入输出路径和跳过文件：
///user/joe/wordcount/input /user/joe/wordcount/output -skip /user/joe/wordcount/patterns.txt
if (!(remainingArgs.length != 2 || remainingArgs.length != 4)) {
System.err
.println(“Usage: wordcount <in> <out> [-skip skipPatternFile]”);
System.exit(2);
}
Job job = Job.getInstance(conf, “word count”);
job.setJarByClass(WordCount2.class);
job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
List<String> otherArgs = new ArrayList<String>(); // 除了 -skip 以外的其它参数
for (int i = 0; i < remainingArgs.length; ++i) {
if (“-skip”.equals(remainingArgs[i])) {
job.addCacheFile(new Path(remainingArgs[++i]).toUri()); // 将
// -skip
// 后面的参数，即 skip 模式文件的 url，加入本地化缓存中
job.getConfiguration().setBoolean(“wordcount.skip.patterns”,
true); // 这里设置的 wordcount.skip.patterns 属性，在 mapper 中使用
} else {
otherArgs.add(remainingArgs[i]); // 将除了 -skip
// 以外的其它参数加入 otherArgs 中
}
}
FileInputFormat.addInputPath(job, new Path(otherArgs.get(0))); // otherArgs 的第一个参数是输入路径
FileOutputFormat.setOutputPath(job, new Path(otherArgs.get(1))); // otherArgs 的第二个参数是输出路径
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}