2014-11-17
这里的“词频”是指单词出现的频数，也就是次数。
在 Hadoop 2.4 实现单词计数 中已经论述了如何进行单词计数，其处理的结果保存在HDFS中的/output目录下，其中有一文件_SUCCESS是空文件（因为是空文件，所以可以忽略），表示这个JOB成功执行了。 另外一个文件是part-r-00000，r代表着这个文件是reduce的结果。
现在对/output中的文件进行处理，获取最大词频。创建java文件MaxNum.java，内容如下：
import java.io.IOException;
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;
public class MaxNum {
public static class MaxNumMapper extends
Mapper<Object, Text, IntWritable, IntWritable> {
private final static IntWritable onlyKey = new IntWritable(1);
public void map(Object key, Text value, Context context)
throws IOException, InterruptedException {
String numStr = value.toString().split("\t")[1];
context.write(onlyKey, new IntWritable(Integer.parseInt(numStr)));
}
}
public static class MaxNumReducer extends
Reducer<IntWritable, IntWritable, Text, IntWritable> {
private final static Text onlyKey = new Text("max");
private IntWritable result = new IntWritable();
public void reduce(IntWritable key, Iterable<IntWritable> values,
Context context) throws IOException, InterruptedException {
int max_num = 0;
for (IntWritable val : values) {
if ( max_num < val.get()) {
max_num = val.get();
}
}
result.set(max_num);
context.write(onlyKey, result);
}
}
public static void main(String[] args) throws Exception {
Configuration conf2 = new Configuration();
Job job2 = Job.getInstance(conf2, "get max number");
job2.setJarByClass(WordCountAndMaxNum.class);
job2.setMapperClass(MaxNumMapper.class);
job2.setMapOutputKeyClass(IntWritable.class);
job2.setMapOutputValueClass(IntWritable.class);
// job2.setCombinerClass(MaxNumReducer.class);
job2.setReducerClass(MaxNumReducer.class);
job2.setOutputKeyClass(Text.class);
job2.setOutputValueClass(IntWritable.class);
FileInputFormat.setInputPaths(job2, "/output");
FileOutputFormat.setOutputPath(job2, new Path("/output2"));
job2.waitForCompletion(true);
}
}
注意，在main()函数中job2.setCombinerClass(MaxNumReducer.class);被注释掉了，如果不注释掉，在运行时会产生这样一个错误：
Error: java.io.IOException: wrong key class: class org.apache.hadoop.io.Text is not class org.apache.hadoop.io.IntWritable
原因是这样的。Combiner过程发生在Map和Reduce之间，它是一个微型的Reduce（一个Combiner Task处理的数据量较小）。在设置Combiner后，意味着这整个JOB有两次Reduce，第一次是Combiner TASK调用MaxNumReducer类，输出的键值类型是<Text, IntWritable>，该输出会作为第二次Reduce的输入；第二次是Reduce TASK调用MaxNumReducer类，要求输入的键值类型为IntWritable, IntWritable，由此便产生了类型的冲突。
如果一定要加上Combiner，有两个方案： 1、修改MaxNumReducer类； 2、再添加一个继承了Reducer的类供Combiner单独使用。
下面我们将单词计数和获取最大词频整合在一起，创建WordCountAndMaxNum.java，内容如下：
import java.io.IOException;
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;
public class WordCountAndMaxNum {
public static class TokenizerMapper extends
Mapper<Object, Text, Text, IntWritable> {
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
public void map(Object key, Text value, Context context)
throws IOException, InterruptedException {
StringTokenizer itr = new StringTokenizer(value.toString());
while (itr.hasMoreTokens()) {
word.set(itr.nextToken());
context.write(word, one);
}
}
}
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);
}
}
public static class MaxNumMapper extends
Mapper<Object, Text, IntWritable, IntWritable> {
private final static IntWritable onlyKey = new IntWritable(1);
public void map(Object key, Text value, Context context)
throws IOException, InterruptedException {
String numStr = value.toString().split("\t")[1];
context.write(onlyKey, new IntWritable(Integer.parseInt(numStr)));
}
}
public static class MaxNumReducer extends
Reducer<IntWritable, IntWritable, Text, IntWritable> {
private final static Text onlyKey = new Text("max");
private IntWritable result = new IntWritable();
public void reduce(IntWritable key, Iterable<IntWritable> values,
Context context) throws IOException, InterruptedException {
int max_num = 0;
for (IntWritable val : values) {
if ( max_num < val.get()) {
max_num = val.get();
}
}
result.set(max_num);
context.write(onlyKey, result);
}
}
public static void main(String[] args) throws Exception {
Configuration conf1 = new Configuration();
Job job1 = Job.getInstance(conf1, "word count");
job1.setJarByClass(WordCountAndMaxNum.class);
job1.setMapperClass(TokenizerMapper.class);
job1.setMapOutputKeyClass(Text.class); //!
job1.setMapOutputValueClass(IntWritable.class); //!
job1.setCombinerClass(IntSumReducer.class);
job1.setReducerClass(IntSumReducer.class);
job1.setOutputKeyClass(Text.class);
job1.setOutputValueClass(IntWritable.class);
FileInputFormat.setInputPaths(job1, "/input");
FileOutputFormat.setOutputPath(job1, new Path("/output"));
job1.waitForCompletion(true);
// --
Configuration conf2 = new Configuration();
Job job2 = Job.getInstance(conf2, "get max number");
job2.setJarByClass(WordCountAndMaxNum.class);
job2.setMapperClass(MaxNumMapper.class);
job2.setMapOutputKeyClass(IntWritable.class);
job2.setMapOutputValueClass(IntWritable.class);
// job2.setCombinerClass(MaxNumReducer.class);
job2.setReducerClass(MaxNumReducer.class);
job2.setOutputKeyClass(Text.class);
job2.setOutputValueClass(IntWritable.class);
FileInputFormat.setInputPaths(job2, "/output");
FileOutputFormat.setOutputPath(job2, new Path("/output2"));
job2.waitForCompletion(true);
}
}
仍然处理 Hadoop 2.4 实现单词计数中使用的文本，结果如下：
zsh >> $HADOOP_PREFIX/bin/hadoop fs -cat /output2/part-r-00000
max 3
我在编码过程中遇到过这样一个问题：
java.io.IOException: Type mismatch in key from map: expected org.apache.hadoop.io.Text, received org.apache.hadoop.io.LongWritable