数据源

A
B
C
D
Z

要实现的输出

Z
D
B
C
A

看字符顺序，其实什么也没有，只是按照后面的数字进行一次倒序排序，实现思路，1利用hadoop自带的排序功能，2.KV互换

实现代码

public class SVJob {
public static void main(String[] args) throws IOException,
InterruptedException, ClassNotFoundException {
Configuration conf = new Configuration();
conf.set("mapred.job.tracker", "192.168.9.181:9001");
String[] ars = new String[] {
"hdfs://192.168.9.181:9000/user/hadoop/input/examples/SortByValue/",
"hdfs://192.168.9.181:9000/user/hadoop/output/examples/SortByValue" };
String[] otherArgs = new GenericOptionsParser(conf, ars)
.getRemainingArgs();
if (otherArgs.length != 2) {
System.err.println("SortByValue: ");
System.exit(2);
}

    Job job = new Job(conf, "SortByValue");  
    job.setJarByClass(SVJob.class);  
    job.setMapperClass(SVMapper.class);  
    job.setReducerClass(SVReducer.class);

    job.setMapOutputKeyClass(IntWritable.class);  
    job.setMapOutputValueClass(Text.class);

    job.setOutputKeyClass(Text.class);  
    job.setOutputValueClass(IntWritable.class);

    job.setSortComparatorClass(IntWritableDecreasingComparator.class);

    FileInputFormat.addInputPath(job, new Path(otherArgs\[0\]));  
    FileOutputFormat.setOutputPath(job, new Path(otherArgs\[1\]));  
    System.exit(job.waitForCompletion(true) ? 0 : 1);  
}  

}

public class SVMapper extends Mapper {
protected void map(Object key, Text value, Context context)
throws IOException, InterruptedException {
String line = value.toString();
String[] keyValueStrings = line.split("\t");
if(keyValueStrings.length != 2)
{
//新手，不知道怎么记录日志，也不清楚怎么退出 各位大神如果知道请通知我，谢谢
System.err.println("string format error!!!!!");
return;
}
int outkey = Integer.parseInt(keyValueStrings[1]);
String outvalue = keyValueStrings[0];
context.write(new IntWritable(outkey), new Text(outvalue));
}
}

public class SVReducer extends Reducer {
protected void reduce(IntWritable key, Iterable values,Context context)throws IOException, InterruptedException {
for(Text value : values){
context.write(value, key);
}
}
}

因为我们要实现倒序排序要有自定义的排序方法

public class IntWritableDecreasingComparator extends Comparator {
@SuppressWarnings("rawtypes")
public int compare( WritableComparable a,WritableComparable b){
return -super.compare(a, b);
}
public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
return -super.compare(b1, s1, l1, b2, s2, l2);
}
}

这样就完成了，可以自定义排序了