我知道sortcomparator是用来按键对Map输出进行排序的。为了更好地理解mapreduce框架,我编写了一个自定义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)
我的自定义排序比较类看起来不错。Map完成后,mysortcomparator2的compare方法应该接收“text”键作为输入,因为我返回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
事先谢谢,我会很感激你的帮助。
3条答案
按热度按时间omtl5h9j1#
您还需要实现/重写此方法:
我认为您的比较器是以这样一种方式构造的:超级实现中提到的变量是空的(这是为支持排序而调用的方法,而不是您上面编写的方法)。这就是为什么会出现空指针异常。通过使用不使用变量的实现重写方法,可以避免异常。
wljmcqd82#
实际上,mysortcomparator2构造函数有一个问题。代码应该是
其中第一个参数是您的键类,第二个参数的值
WritableComparator
以这样的方式示例化WritableComparator.compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2)
可以调用MySortComparator2.compare(WritableComparable a, WritableComparable b)
ni65a41a3#
正如chrisgerken所说,在扩展writeablecomparator或实现rawcomarator而不是writeablecomparator时,需要重写这个方法。
正如您所说的,您不希望看到要进行排序,但是如果您返回0,这意味着每次mapreduce尝试排序/比较时,它都会将每个键视为相同的东西,因此,您将只收到一个键,即值对,它将是map任务中第一个先完成的键,以及输入文件中具有字数的值。希望你明白我的意思。如果你的意见是这样的
你的减产将是
因为它假设所有的东西都是同一把钥匙。我希望这有帮助。