我知道SortComparator是用来按键对映射输出进行排序的。为了更好地理解MapReduce框架，我编写了一个定制的SortComparator。这是我的WordCount类和自定义的SortComparator类。
package bananas;
import java.io.FileWriter;
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.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class WordCount {
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 MyPartitoner extends Partitioner<Text, IntWritable>{
@Override
public int getPartition(Text key, IntWritable value, int numPartitions) {
return Math.abs(key.hashCode()) % numPartitions;
}
}
public static class MySortComparator2 extends WritableComparator{
protected MySortComparator2() {
super();
}
@SuppressWarnings({ "rawtypes" })
@Override
public int compare(WritableComparable w1,WritableComparable w2){
return 0;
}
}
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "word count");
job.setJarByClass(WordCount.class);
job.setSortComparatorClass(MySortComparator2.class);
job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
但是当我执行这个时，我得到了这个错误
Error: java.lang.NullPointerException
at org.apache.hadoop.io.WritableComparator.compare(WritableComparator.java:157)
at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.compare(MapTask.java:1265)
at org.apache.hadoop.util.QuickSort.fix(QuickSort.java:35)
at org.apache.hadoop.util.QuickSort.sortInternal(QuickSort.java:87)
at org.apache.hadoop.util.QuickSort.sort(QuickSort.java:63)
at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.sortAndSpill(MapTask.java:1593)
at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.flush(MapTask.java:1482)
at org.apache.hadoop.mapred.MapTask$NewOutputCollector.close(MapTask.java:720)
at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:790)
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:341)
at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:163)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:415)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1628)
at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:158)
我的自定义SortComparator类看起来很好。映射完成后，MySortComparator2的比较方法应该接收"文本"键作为输入，因为我返回0，没有排序将完成。这就是我期望看到/观察到的。我遵循这些教程
http://codingjunkie.net/secondary-sort/http://blog.zaloni.com/secondary-sorting-in-hadoop http://www.bigdataspeak.com/2013/02/hadoop-how-to-do-secondary-sort-on_25.html提前表示感谢。
实际上，MySortComparator2构造函数有一个问题。代码应该看起来像
protected MySortComparator2() {
super(Text.class, true);
}
其中第一个参数是您的关键类，第二个参数的值确保WritableComparator以WritableComparator.compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2)可以调用MySortComparator2.compare(WritableComparable a, WritableComparable b)的方式实例化
您还需要实现/重写此方法:
public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
// per your desired no-sort logic
return 0;
}
我认为你的比较器是以这样一种方式构造的，即超级实现中提到的变量为null(这是为了支持排序而调用的方法-而不是你上面写的方法)。这就是为什么你会得到空指针异常。通过使用不使用变量的实现重写方法，您可以避免异常。
正如Chris Gerken所说，你需要在扩展WritableComparator时重写这个方法，或者实现RawComparator而不是WritableComparator。
public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
return 0;
}
，正如你说的，你想看到没有排序要做，但如果你返回0，这意味着每次MapReduce尝试排序/比较它看到每个键作为相同的事情，所以，你将只收到一个键，值对，这将是第一个键在地图任务，首先得到完成的值与输入文件中的单词数。希望你明白我在说什么。如果你的输入是这样的
why are rockets cylindrical
您的reduce输出将是
why 4
，因为它假设所有内容都是相同的键。