在Hadoop中的例子TeraSort，就是一個利用mapredue進行排序的例子。本文參考並簡化了這個例子：
排序的基本思想是利用了mapreduce的自動排序功能，在hadoop中，從map到reduce階段，map出來的結構會按照各個key按照 hash值分配到各個reduce中，其中，在reduce中所有的key都是有序的了。如果使用一個reduce，那麼我們直接將他output出來就 行了，但是這不能夠體現分布式的好處，所以，我們還是要用多個reduce來跑。
比方說我們有1000個1-10000的數據，跑10個ruduce任務， 如果我們運行進行partition的時候，能夠將在1-1000中數據的分配到第一個reduce中，1001-2000的數據分配到第二個 reduce中，以此類推。即第n個reduce所分配到的數據全部大於第n-1個reduce中的數據。這樣，每個reduce出來之後都是有序的了， 我們只要cat所有的輸出文件，變成一個大的文件，就都是有序的了。
基本思路就是這樣，但是現在有一個問題，就是數據的區間如何劃分，在數據量大，還有我們並不清楚數據分布的情況下。一個比較簡單的方法就是采樣，假如有一 億的數據，我們可以對數據進行采樣，如取10000個數據采樣，然後對采樣數據分區間。在Hadoop中，patition我們可以用 TotalOrderPartitioner替換默認的分區。然後將采樣的結果傳給他，就可以實現我們想要的分區。在采樣時，我們可以使用hadoop的 幾種采樣工具，RandomSampler,InputSampler,IntervalSampler。
這樣，我們就可以對利用分布式文件系統進行大數據量的排序了，我們也可以重寫Partitioner類中的compare函數，來定義比較的規則，從而可以實現字符串或其他非數字類型的排序，也可以實現二次排序乃至多次排序。
參考：《Hadoop權威指南》裡面有詳細的講解 下載見 http://www.linuxidc.com/Linux/2012-07/65972.htm
CxfInputFormat.java
package com.alibaba.cxf.sort;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.LineRecordReader;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;
public class CxfInputFormat extends FileInputFormat<IntWritable,Text>{
@Override
public RecordReader<IntWritable, Text> getRecordReader(InputSplit split,
JobConf job, Reporter reporter) throws IOException {
return new CxfRecordReader(job, (FileSplit) split);
}
class CxfRecordReader implements RecordReader<IntWritable,Text> {
private LineRecordReader in;
private LongWritable junk = new LongWritable();
private Text line = new Text();
private int KEY_LENGTH = 10;
public CxfRecordReader(JobConf job,FileSplit split) throws IOException{
in = new LineRecordReader(job, split);
}
@Override
public void close() throws IOException {
in.close();
}
@Override
public IntWritable createKey() {
return new IntWritable();
}
@Override
public Text createValue() {
return new Text();
}
@Override
public long getPos() throws IOException {
return in.getPos();
}
@Override
public float getProgress() throws IOException {
return in.getProgress();
}
@Override
public boolean next(IntWritable key, Text value) throws IOException {
if (in.next(junk, line)) {
if (line.getLength() < KEY_LENGTH) {
key.set(Integer.parseInt(line.toString()));
value = new Text();
// value.clear();
} else {
byte[] bytes = line.getBytes();
key.set(Integer.parseInt(new String(bytes).substring(0, KEY_LENGTH)));
value = new Text();
}
return true;
} else {
return false;
}
}
}
}
SortByMapReduce.java
package com.alibaba.cxf.sort;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapred.lib.InputSampler;
import org.apache.hadoop.mapred.lib.TotalOrderPartitioner;
public class SortByMapReduce {
/**
* @param args
* @throws URISyntaxException
* @throws IOException
*/
public static void main(String[] args) throws IOException, URISyntaxException {
runJob(args);
}
private static void runJob(String[] args) throws IOException, URISyntaxException {
JobConf conf = new JobConf(SortByMapReduce.class);
FileInputFormat.setInputPaths(conf, new Path(args[0]));
FileOutputFormat.setOutputPath(conf, new Path(args[1]));
conf.setJobName(”SortByMapReduce”);
conf.setInputFormat(CxfInputFormat.class);
conf.setOutputKeyClass(IntWritable.class);
conf.setOutputFormat(TextOutputFormat.class);
conf.setNumReduceTasks(5);
conf.setPartitionerClass(TotalOrderPartitioner.class);
InputSampler.RandomSampler<IntWritable, NullWritable> sampler =
new InputSampler.RandomSampler<IntWritable, NullWritable>(0.1,10000,10);
Path input = FileInputFormat.getInputPaths(conf)[0];
input = input.makeQualified(input.getFileSystem(conf));
Path partitionFile = new Path(input,”_partitions”);
TotalOrderPartitioner.setPartitionFile(conf, partitionFile);
InputSampler.writePartitionFile(conf, sampler);
URI partitionURI = new URI(partitionFile.toString() + “#_partitions”);
DistributedCache.addCacheFile(partitionURI, conf);
DistributedCache.createSymlink(conf);
JobClient.runJob(conf);
}
}
更多Hadoop相關信息見Hadoop 專題頁面 http://www.linuxidc.com/topicnews.aspx?tid=13
