本文轉(zhuǎn)自：http://www.alidw.com/?p=1420
在hadoop中的例子TeraSort，就是一個(gè)利用mapredue進(jìn)行排序的例子。本文參考并簡(jiǎn)化了這個(gè)例子：
排序的基本思想是利用了mapreduce的自動(dòng)排序功能，在hadoop中，從map到reduce階段，map出來(lái)的結(jié)構(gòu)會(huì)按照各個(gè)key按照 hash值分配到各個(gè)reduce中，其中， 在reduce中所有的key都是有序的了 。如果使用一個(gè)reduce，那么我們直接將他output出來(lái)就 行了，但是這不能夠體現(xiàn)分布式的好處，所以，我們還是要用多個(gè)reduce來(lái)跑。
比方說(shuō)我們有1000個(gè)1-10000的數(shù)據(jù)，跑10個(gè)ruduce任務(wù)， 如果我們運(yùn)行進(jìn)行partition的時(shí)候，能夠?qū)⒃?-1000中數(shù)據(jù)的分配到第一個(gè)reduce中 ，1001-2000的數(shù)據(jù)分配到第二個(gè) reduce中，以此類(lèi)推。即第n個(gè)reduce所分配到的數(shù)據(jù)全部大于第n-1個(gè)reduce中的數(shù)據(jù)。這樣，每個(gè)reduce出來(lái)之后都是有序的了， 我們只要cat所有的輸出文件， 變成一個(gè)大的文件，就都是有序的了 。
基本思路就是這樣，但是現(xiàn)在有一個(gè)問(wèn)題，就是 數(shù)據(jù)的區(qū)間如何劃分 ，在數(shù)據(jù)量大，還有我們并不清楚數(shù)據(jù)分布的情況下。一個(gè)比較簡(jiǎn)單的方法就是采樣，假如有一 億的數(shù)據(jù)，我們可以對(duì)數(shù)據(jù)進(jìn)行采樣，如取10000個(gè)數(shù)據(jù)采樣，然后對(duì)采樣數(shù)據(jù)分區(qū)間。在Hadoop中，patition我們可以用 TotalOrderPartitioner替換默認(rèn)的分區(qū)。然后將采樣的結(jié)果傳給他，就可以實(shí)現(xiàn)我們想要的分區(qū)。在采樣時(shí)，我們可以使用hadoop的 幾種采樣工具，RandomSampler,InputSampler,IntervalSampler。
這樣，我們就可以對(duì)利用分布式文件系統(tǒng)進(jìn)行大數(shù)據(jù)量的排序了，我們也可以重寫(xiě)Partitioner類(lèi)中的compare函數(shù)，來(lái)定義比較的規(guī)則，從而可以實(shí)現(xiàn)字符串或其他非數(shù)字類(lèi)型的排序，也可以實(shí)現(xiàn)二次排序乃至多次排序。
參考：《Hadoop權(quán)威指南》里面有詳細(xì)的講
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
}
更多文章、技術(shù)交流、商務(wù)合作、聯(lián)系博主
微信掃碼或搜索：z360901061
微信掃一掃加我為好友
QQ號(hào)聯(lián)系： 360901061
您的支持是博主寫(xiě)作最大的動(dòng)力，如果您喜歡我的文章，感覺(jué)我的文章對(duì)您有幫助，請(qǐng)用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧，狠狠點(diǎn)擊下面給點(diǎn)支持吧，站長(zhǎng)非常感激您！手機(jī)微信長(zhǎng)按不能支付解決辦法：請(qǐng)將微信支付二維碼保存到相冊(cè)，切換到微信，然后點(diǎn)擊微信右上角掃一掃功能，選擇支付二維碼完成支付。
【本文對(duì)您有幫助就好】元
