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

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

? ?欢迎加入Hadoop超级群:?180941958

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

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

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

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

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

?

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

????? Hadoop MapReduce 学习笔记(八) MapReduce实现类似SQL的order by/排序写法会触发内存溢出,因为将数据都放到一个内存容器中.这是我一开始所想到的,后来触发了OOM.于是想到了借用map输出的自己的排序,网上找资料以及hadoop/examples下也是这种写法.代码如下:

?

package com.guoyun.hadoop.mapreduce.study;import java.io.IOException;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.FileSystem;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Job;import org.apache.hadoop.mapreduce.Mapper;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;/** * 对具有多列数据中的一列进行从小到大排序，并返回结果 * 如果想实现从大到小，或者多列数据排序,请自己实现 @WritableComparable * 需要实现对多列排序，例如SQL： * SELECT ＊ FROM TABLE ORDER BY COL1 ASC，COL2 ASC 请查看 ＠OrderByMultiMapReduceTest */public class OrderBySingleMapReduceFixTest extends MyMapReduceMultiColumnTest {    public static final Logger log=LoggerFactory.getLogger(OrderBySingleMapReduceFixTest.class);    public OrderBySingleMapReduceFixTest(long dataLength) throws Exception {    super(dataLength);    // TODO Auto-generated constructor stub  }  public OrderBySingleMapReduceFixTest(String outputPath) throws Exception {    super(outputPath);    // TODO Auto-generated constructor stub  }  public OrderBySingleMapReduceFixTest(long dataLength, String inputPath,      String outputPath) throws Exception {    super(dataLength, inputPath, outputPath);    // TODO Auto-generated constructor stub  }    /**   * Map,to get the source datas   */  private static class MyMapper extends Mapper<LongWritable,Text,LongWritable,Text>{    private final LongWritable writeKey=new LongWritable(0);    private Text writeValue=new Text();        @Override    protected void map(LongWritable key, Text value, Context context)        throws IOException, InterruptedException {      log.debug("begin to map");      String[] splits=null;            try {        splits=value.toString().split("\\t");        if(splits!=null&&splits.length==2){          writeValue.set(splits[0]);          writeKey.set(Long.parseLong(splits[1].trim()));        }      } catch (NumberFormatException e) {        log.error("map error:"+e.getMessage());      }            context.write(writeKey, writeValue);    }  }    private static class MyReducer     extends Reducer<LongWritable,Text,LongWritable,Text>{    @Override    protected void reduce(LongWritable key, Iterable<Text> values,        Context context) throws IOException, InterruptedException {      for(Text value:values){        context.write(key, value);      }    }      }    /**   * @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_Fix";    String output="testDatas/mapreduce/MROutput_Single_OrderBy_Fix";            try {      mapReduceTest=new OrderBySingleMapReduceFixTest(100000,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(OrderBySingleMapReduceFixTest.class);      job.setMapOutputKeyClass(LongWritable.class);      job.setMapOutputValueClass(Text.class);      job.setOutputKeyClass(LongWritable.class);      job.setOutputValueClass(Text.class);      job.setMapperClass(MyMapper.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();    }      }}