[转载]使用Hadoop MapReduce 进行排序

ab520025520025

　　      本文转自：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{
20  @Override
21  public RecordReader getRecordReader(InputSplit split,
22    JobConf job, Reporter reporter) throws IOException {
23   return new CxfRecordReader(job, (FileSplit) split);
24  }
25  class CxfRecordReader implements RecordReader {
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 sampler =
122    new InputSampler.RandomSampler(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 }
　　
解