本文转自：http://www.alidw.com/?p=1420
在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权威指南》里面有详细的讲
1
CxfInputFormat.java
2
3
package
com.alibaba.cxf.sort;
4
5
import
java.io.IOException;
6
7
import
org.apache.hadoop.io.IntWritable;
8
import
org.apache.hadoop.io.LongWritable;
9
import
org.apache.hadoop.io.NullWritable;
10
import
org.apache.hadoop.io.Text;
11
import
org.apache.hadoop.mapred.FileInputFormat;
12
import
org.apache.hadoop.mapred.FileSplit;
13
import
org.apache.hadoop.mapred.InputSplit;
14
import
org.apache.hadoop.mapred.JobConf;
15
import
org.apache.hadoop.mapred.LineRecordReader;
16
import
org.apache.hadoop.mapred.RecordReader;
17
import
org.apache.hadoop.mapred.Reporter;
18
19
public
class
CxfInputFormat
extends
FileInputFormat<IntWritable,Text>{
20
@Override
21
public
RecordReader<IntWritable, Text> getRecordReader(InputSplit split,
22
JobConf job, Reporter reporter)
throws
IOException {
23
return
new
CxfRecordReader(job, (FileSplit) split);
24
}
25
class
CxfRecordReader
implements
RecordReader<IntWritable,Text> {
26
27
private
LineRecordReader in;
28
private
LongWritable junk =
new
LongWritable();
29
private
Text line =
new
Text();
30
private
int
KEY_LENGTH = 10;
31
public
CxfRecordReader(JobConf job,FileSplit split)
throws
IOException{
32
in =
new
LineRecordReader(job, split);
33
}
34
@Override
35
public
void
close()
throws
IOException {
36
in.close();
37
}
38
@Override
39
public
IntWritable createKey() {
40
return
new
IntWritable();
41
}
42
@Override
43
public
Text createValue() {
44
45
return
new
Text();
46
}
47
@Override
48
public
long
getPos()
throws
IOException {
49
50
return
in.getPos();
51
}
52
@Override
53
public
float
getProgress()
throws
IOException {
54
55
return
in.getProgress();
56
}
57
@Override
58
public
boolean
next(IntWritable key, Text value)
throws
IOException {
59
if
(in.next(junk, line)) {
60
if
(line.getLength() < KEY_LENGTH) {
61
key.set(Integer.parseInt(line.toString()));
62
value =
new
Text();
63
//
value.clear();
64
}
else
{
65
byte
[] bytes = line.getBytes();
66
key.set(Integer.parseInt(
new
String(bytes).substring(0, KEY_LENGTH)));
67
value =
new
Text();
68
}
69
return
true
;
70
}
else
{
71
return
false
;
72
}
73
}
74
}
75
}
76
77
78
79
SortByMapReduce.java
80
81
package
com.alibaba.cxf.sort;
82
83
import
java.io.IOException;
84
import
java.net.URI;
85
import
java.net.URISyntaxException;
86
import
org.apache.hadoop.filecache.DistributedCache;
87
import
org.apache.hadoop.fs.Path;
88
import
org.apache.hadoop.io.IntWritable;
89
import
org.apache.hadoop.io.NullWritable;
90
import
org.apache.hadoop.mapred.FileInputFormat;
91
import
org.apache.hadoop.mapred.FileOutputFormat;
92
import
org.apache.hadoop.mapred.JobClient;
93
import
org.apache.hadoop.mapred.JobConf;
94
import
org.apache.hadoop.mapred.TextOutputFormat;
95
import
org.apache.hadoop.mapred.lib.InputSampler;
96
import
org.apache.hadoop.mapred.lib.TotalOrderPartitioner;
97
public
class
SortByMapReduce {
98
99
/**
100
*
@param
args
101
*
@throws
URISyntaxException
102
*
@throws
IOException
103
*/
104
public
static
void
main(String[] args)
throws
IOException, URISyntaxException {
105
runJob(args);
106
}
107
108
private
static
void
runJob(String[] args)
throws
IOException, URISyntaxException {
109
110
JobConf conf =
new
JobConf(SortByMapReduce.
class
);
111
112
FileInputFormat.setInputPaths(conf,
new
Path(args[0]));
113
FileOutputFormat.setOutputPath(conf,
new
Path(args[1]));
114
conf.setJobName(”SortByMapReduce”);
115
116
conf.setInputFormat(CxfInputFormat.
class
);
117
conf.setOutputKeyClass(IntWritable.
class
);
118
conf.setOutputFormat(TextOutputFormat.
class
);
119
conf.setNumReduceTasks(5);
120
conf.setPartitionerClass(TotalOrderPartitioner.
class
);
121
InputSampler.RandomSampler<IntWritable, NullWritable> sampler =
122
new
InputSampler.RandomSampler<IntWritable, NullWritable>(0.1,10000,10);
123
124
Path input = FileInputFormat.getInputPaths(conf)[0];
125
input = input.makeQualified(input.getFileSystem(conf));
126
Path partitionFile =
new
Path(input,”_partitions”);
127
TotalOrderPartitioner.setPartitionFile(conf, partitionFile);
128
InputSampler.writePartitionFile(conf, sampler);
129
130
URI partitionURI =
new
URI(partitionFile.toString() + “#_partitions”);
131
DistributedCache.addCacheFile(partitionURI, conf);
132
DistributedCache.createSymlink(conf);
133
JobClient.runJob(conf);
134
}
135
}
