本文轉自：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
}
更多文章、技術交流、商務合作、聯系博主
微信掃碼或搜索：z360901061
微信掃一掃加我為好友
QQ號聯系： 360901061
您的支持是博主寫作最大的動力，如果您喜歡我的文章，感覺我的文章對您有幫助，請用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧，狠狠點擊下面給點支持吧，站長非常感激您！手機微信長按不能支付解決辦法：請將微信支付二維碼保存到相冊，切換到微信，然后點擊微信右上角掃一掃功能，選擇支付二維碼完成支付。
【本文對您有幫助就好】元
