/**

* 自己定義的key類應該實現WritableComparable介面 */

public	class IntPair implements WritableComparable<IntPair>{ int first;//第一個成員變數 int second;//第二個成員變數

public void set(int left, int right){ first = left;

second = right;

}

public int getFirst(){ return first;
}

public int getSecond(){ return second;
}

@Override

//反序列化，從流中的二進位制轉換成IntPair public void readFields(DataInput in) throws IOException{

first = in.readInt(); second = in.readInt();
}

@Override

//序列化，將IntPair轉化成使用流傳送的二進位制 public void write(DataOutput out) throws IOException{

out.writeInt(first);

out.writeInt(second);

}

@Override //key的比較

public int compareTo(IntPair o)

{

// TODO Auto‐generated method stub if (first != o.first){

return first < o.first ? -1 : 1; }else if (second != o.second){

return second < o.second ? -1 : 1; }else{

return 0;

}

}

@Override

public int hashCode(){

return first * 157 + second;

}

@Override

public boolean equals(Object right){ if (right == null)

return false; if (this == right) return true;

if (right instanceof IntPair){ IntPair r = (IntPair) right;

return r.first == first && r.second == second; }else{

return false;

}

}

}

第二步：自定義分割槽函式類FirstPartitioner，根據 IntPair 中的first實現分割槽。


/**

* 分割槽函式類。根據first確定Partition。 */

public static class FirstPartitioner extends Partitioner<IntPair, IntWritable>{ @Override

public int getPartition(IntPair key, IntWritable value,int numPartitions){ return Math.abs(key.getFirst() * 127) % numPartitions;
}

}

第三步：自定義 SortComparator 實現 IntPair 類中的first和second排序。本課程中沒

有使用這種方法，而是使用 IntPair 中的compareTo()方法實現的。

第四步：自定義 GroupingComparator 類，實現分割槽內的資料分組。

/** *繼承WritableComparator */

public static class GroupingComparator extends WritableComparator{ protected GroupingComparator(){
super(IntPair.class, true);

}

@Override

//Compare two WritableComparables.

public int compare(WritableComparable w1, WritableComparable w2){ IntPair ip1 = (IntPair) w1;
IntPair ip2 = (IntPair) w2; int l = ip1.getFirst(); int r = ip2.getFirst();

return l == r ? 0 : (l < r ? -1 : 1);

}

}

第五步：編寫 MapReduce 主程式實現二次排序。

	
import	java.io.DataInput;
import	java.io.DataOutput;
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.LongWritable;
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.input.TextInputFormat;
import	org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import	org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
public	class SecondarySort{
//	自定義map
public static class Map extends Mapper<LongWritable, Text, IntPair, IntWritable>{
	private final IntPair intkey = new IntPair();
	private final IntWritable intvalue = new IntWritable();
	public void map(LongWritable key, Text value, Context context) throws IOException, Interrupted
	String line = value.toString();
	StringTokenizer tokenizer = new StringTokenizer(line);
	int left = 0;
	int right = 0;
	if (tokenizer.hasMoreTokens()){
	left = Integer.parseInt(tokenizer.nextToken());
	if (tokenizer.hasMoreTokens())
	right = Integer.parseInt(tokenizer.nextToken());
	intkey.set(left, right);
	intvalue.set(right);
	context.write(intkey, intvalue);
	}
	}
}	
//	自定義reduce

public static class Reduce extends Reducer< IntPair, IntWritable, Text, IntWritable>{ private final Text left = new Text();

public void reduce(IntPair key, Iterable< IntWritable> values,Context context) throws IOExcepti left.set(Integer.toString(key.getFirst()));
for (IntWritable val : values){ context.write(left, val);
}

}

}

/**

* @param args */
public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundExce // TODO Auto‐generated method stub

Configuration conf = new Configuration();

Job job = new Job(conf, "secondarysort"); job.setJarByClass(SecondarySort.class);

FileInputFormat.setInputPaths(job, new Path(args[0]));//輸入路徑

FileOutputFormat.setOutputPath(job, new Path(args[1]));//輸出路徑

job.setMapperClass(Map.class);// Mapper job.setReducerClass(Reduce.class);// Reducer

job.setPartitionerClass(FirstPartitioner.class);// 分割槽函式//job.setSortComparatorClass(KeyComparator.Class);//本課程並沒有自定義SortComparator，而是使用In job.setGroupingComparatorClass(GroupingComparator.class);// 分組函式




job.setMapOutputKeyClass(IntPair.class);

job.setMapOutputValueClass(IntWritable.class);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

job.setInputFormatClass(TextInputFormat.class);

job.setOutputFormatClass(TextOutputFormat.class);

System.exit(job.waitForCompletion(true) ? 0 : 1);

}

}
 

MapReduce 的二次排序的原理和實現OK!