首页 / 操作系统 / Linux / 使用Hadoop MapReduce 进行排序

在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.htmCxfInputFormat.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