hadoop中map/reduce数据排序实战

需求描述

对输入文件中数据进行排序。输入文件中的每行内容均为一个数字，即一个数据。
要求在输出中每行有两个间隔的数字，其中，第一个代表原始数据在原始数据集中的位次，第二个代表原始数据。

设计思考

这个实例仅仅要求对输入数据进行排序，熟悉MapReduce过程的读者会很快想到在MapReduce过程中就有排序，是否可以利用这个默认的排序，这个实例仅仅要求对输入数据进行排序，熟悉MapReduce过程的读者会很快想到在MapReduce过程中就有排序，是否可以利用这个默认的排序， 而不需要自己再实现具体的排序呢？答案是肯定的。
但是在使用之前首先需要了解它的默认排序规则。它是按照key值进行排序的，如果key为封装int的IntWritable类型，那么MapReduce按照数字大小对key排序， 如果key为封装为String的Text类型，那么MapReduce按照字典顺序对字符串排序。
了解了这个细节，我们就知道应该使用封装int的IntWritable型数据结构了。也就是在map中将读入的数据转化成 IntWritable型，然后作为key值输出（value任意）。 reduce拿到<key，value-list>之后，将输入的 key作为value输出，并根据value-list中元素的个数决定输出的次数。 输出的key（即代码中的linenum）是一个全局变量，它统计当前key的位次。需要注意的是这个程序中没有配置Combiner，也就是在MapReduce过程中不使用Combiner。 这主要是因为使用map和reduce就已经能够完成任务了。

原始数据

file1：

2

32

654

32

15

756

65223

file2：

5956

22

650

92

3）file3：

26

54

6

输出：

1 2

2 6

3 15

4 22

5 26

6 32

7 32

8 54

9 92

10 650

11 654

12 756

13 5956

14 65223

map代码

package com.test.hadoop.sort;
 import java.io.IOException;
 import org.apache.hadoop.io.IntWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.Mapper;
 public class IntSortMapper extends Mapper<Object, Text, IntWritable, IntWritable> {
 IntWritable val = new IntWritable(1); 
@Override 
protected void map(Object key, Text value,Context context) throws IOException, InterruptedException {
 context.write(new IntWritable(Integer.valueOf(value.toString())), val);
 }
}

reduce代码

package com.test.hadoop.sort;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;
public class IntSortReducer extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {
    IntWritable num = new IntWritable(1);
    @Override 
    protected void reduce(IntWritable key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {
 for(IntWritable tmp: values){
     context.write(num, tmp);
     num = new IntWritable(num.get()+1);
 }
 }
 }

main代码

package com.test.hadoop.sort;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.test.hadoop.join.two.UserJob;

public class IntSortJob extends Configuration implements Tool, Runnable {

   private String inputPath = null;
   private String outputPath = null;

   public IntSortJob(String inputPath,String outputPath){
      this.inputPath = inputPath;
      this.outputPath = outputPath;
   }
   public IntSortJob(){}

   @Override
   public Configuration getConf() {
      // TODO Auto-generated method stub
      return null;
   }

   @Override
   public void setConf(Configuration arg0) {
      // TODO Auto-generated method stub

   }

   @Override
   public void run() {
      try{
         String[] args = {this.inputPath,this.outputPath};

         start(args);

      }catch (Exception e) {
         e.printStackTrace();
      }

   }

   private void start(String[] args)throws Exception{

      ToolRunner.run(new UserJob(), args);
   }

   @Override
   public int run(String[] args) throws Exception {
      Configuration configuration = new Configuration();
      FileSystem fs = FileSystem.get(configuration);
      fs.delete(new Path(args[1]),true);

      Job job = new Job(configuration,"uniquejob");
      job.setJarByClass(IntSortJob.class);

      job.setMapperClass(IntSortMapper.class);
      job.setReducerClass(IntSortReducer.class);

      job.setOutputKeyClass(IntWritable.class);
      job.setOutputValueClass(IntWritable.class);

      FileInputFormat.addInputPath(job, new Path(args[0]));
      FileOutputFormat.setOutputPath(job, new Path(args[1]));

      boolean success = job.waitForCompletion(true);

      return success?0:1;
   }

}

   package com.test.hadoop.sort;

public class JobMain {

   /**
    * @param args
    */
   public static void main(String[] args) {
      if(args.length==2){
         new Thread(new IntSortJob(args[0],args[1])).start();
      }

   }

}