Hadoop MapReduce 学习笔记(八) MapReduce实现类似SQL的order by/排序

? ? 本博客属原创文章,转载请注明出处:http://guoyunsky.iteye.com/blog/1235945

?????? 本博客已迁移到本人独立博客:http://www.yun5u.com/articles/hadoop-mapreduce-sql-order-by-sort.html

?

?????? 请先阅读:??????????

?????????? 1.Hadoop MapReduce 学习笔记(一) 序言和准备

?????????? 2.Hadoop MapReduce 学习笔记(二) 序言和准备 2

?

??? 下一篇:Hadoop MapReduce 学习笔记(九) MapReduce实现类似SQL的order by/排序 正确写法

?

??? 排序是很重要的一个环节,类似SQL中的SELECT * FROM TABLE ORDER BY ID,如何用MapReduce实现呢?

?

package com.guoyun.hadoop.mapreduce.study;import java.io.IOException;import java.util.PriorityQueue;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.FileSystem;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.NullWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Job;import org.apache.hadoop.mapreduce.Reducer;import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import org.slf4j.Logger;import org.slf4j.LoggerFactory;/** * 对多列数据中某一列进行从小到大排序，然后返回所有数据,类似SQL： * SELECT ＊ FROM TABLE ORDER BY ID ASC； * 但这是一个错误的写法,会内存溢出,避免内存溢出请查看:＠OrderByMapReduceFixTest */public class OrderBySingleMapReduceTest extends MyMapReduceMultiColumnTest {    public static final Logger log=LoggerFactory.getLogger(OrderBySingleMapReduceTest.class);    public OrderBySingleMapReduceTest(long dataLength) throws Exception {    super(dataLength);    // TODO Auto-generated constructor stub  }  public OrderBySingleMapReduceTest(String outputPath) throws Exception {    super(outputPath);    // TODO Auto-generated constructor stub  }  public OrderBySingleMapReduceTest(long dataLength, String inputPath,      String outputPath) throws Exception {    super(dataLength, inputPath, outputPath);    // TODO Auto-generated constructor stub  }    private static class MyReducer     extends Reducer<Text,MultiColumnWritable,NullWritable,MultiColumnWritable>{    PriorityQueue<MultiColumnWritable> queue=new PriorityQueue<MultiColumnWritable>();    @Override    protected void reduce(Text key, Iterable<MultiColumnWritable> values,        Context context) throws IOException, InterruptedException {      MultiColumnWritable copy=null;      for(MultiColumnWritable value:values){        copy=MultiColumnWritable.copy(value);        queue.add(copy);      }            while(!queue.isEmpty()){        copy=queue.poll();        if(copy!=null){          context.write(NullWritable.get(), copy);        }      }    }      }    /**   * @param args   */  public static void main(String[] args) {    MyMapReduceTest mapReduceTest=null;    Configuration conf=null;    Job job=null;    FileSystem fs=null;    Path inputPath=null;    Path outputPath=null;    long begin=0;    String input="testDatas/mapreduce/MRInput_Single_OrderBy";    String output="testDatas/mapreduce/MROutput_Single_OrderBy";            try {      mapReduceTest=new OrderBySingleMapReduceTest(1000,input,output);            inputPath=new Path(mapReduceTest.getInputPath());      outputPath=new Path(mapReduceTest.getOutputPath());            conf=new Configuration();      job=new Job(conf,"OrderBy");            fs=FileSystem.getLocal(conf);      if(fs.exists(outputPath)){        if(!fs.delete(outputPath,true)){          System.err.println("Delete output file:"+mapReduceTest.getOutputPath()+" failed!");          return;        }      }                  job.setJarByClass(OrderBySingleMapReduceTest.class);      job.setMapOutputKeyClass(Text.class);      job.setMapOutputValueClass(MultiColumnWritable.class);      job.setOutputKeyClass(NullWritable.class);      job.setOutputValueClass(MultiColumnWritable.class);      job.setMapperClass(MultiSupMapper.class);      job.setReducerClass(MyReducer.class);            job.setNumReduceTasks(2);            FileInputFormat.addInputPath(job, inputPath);      FileOutputFormat.setOutputPath(job, outputPath);            begin=System.currentTimeMillis();      job.waitForCompletion(true);            System.out.println("===================================================");      if(mapReduceTest.isGenerateDatas()){        System.out.println("The maxValue is:"+mapReduceTest.getMaxValue());        System.out.println("The minValue is:"+mapReduceTest.getMinValue());      }      System.out.println("Spend time:"+(System.currentTimeMillis()-begin));      // Spend time:13361          } catch (Exception e) {      // TODO Auto-generated catch block      e.printStackTrace();    }      }}