mapreduce的二次排序（字符型）

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.WritableComparable;

public class StringPair implements WritableComparable {
private String name;
private String addr;

public void write(DataOutput out) throws IOException {
out.writeUTF(name);
out.writeUTF(addr);
}

public void readFields(DataInput in) throws IOException {
this.name = in.readUTF();
this.addr = in.readUTF();
}

public int compareTo(StringPair o) {
if (!this.name.equals(o.name)) {
return (int) this.name.compareTo(o.name);
} else {
return (int) this.addr.compareTo(o.addr);
}
}

public String getName() {
return name;
}

public void setName(String name) {
this.name = name;
}

public String getAddr() {
return addr;
}

public void setAddr(String addr) {
this.addr = addr;
}

}

package test;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
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 SecondSort {

public static class MyPatitioner extends Partitioner{

@Override
public int getPartition(StringPair key, Text value, int numPartitions) {

return (key.getName().hashCode() & Integer.MAX_VALUE) % numPartitions;

}

}

public static class MyGroup extends WritableComparator{
public MyGroup(){
// 将要比较的数据类型发给WritableComparator父类，因为欧文继承来自WritableComparator
// 的原因就是想使用WritableComparator中已经定义的一些特性
super(StringPair.class,true) ;
}

@SuppressWarnings("rawtypes")
@Override
public int compare(WritableComparable a, WritableComparable b) {
StringPair o1 = (StringPair)a ;
StringPair o2 = (StringPair)b ;
String first = o1.getName() ;
String second = o2.getName() ;
return first.compareTo(second) ;
}

}

public static class SSmap extends Mapper