map/reduce实现 排序

 import java.io.IOException;

 import org.apache.hadoop.conf.Configuration;

 import org.apache.hadoop.conf.Configured;

 import org.apache.hadoop.fs.Path;

 import org.apache.hadoop.io.IntWritable;

 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.Partitioner;

 import org.apache.hadoop.mapreduce.Reducer;

 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

 import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

 import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

 import org.apache.hadoop.util.Tool;

 import org.apache.hadoop.util.ToolRunner;

 public class Sort extends Configured implements Tool {

     /*

      * 排序

      * 输入格式：每个数据占一行

      * 输出格式：

      * 1 21

      * 2 32

      * 3 62

      * 设计思路：

      * 使用reduce自带的默认排序规则。MapReduce按照key值进行排序。如果Key值为Intwritable类型，则按照数字大小排序

      * 如果key值为Text类型，则按照字典顺序对字符串进行排序。

      * 注意：要重写Partition函数。Reduce排序只能保证自己局部的数据顺序，并不能保证全局的。

      * */

     public static class Map extends Mapper<LongWritable,Text,IntWritable,IntWritable>{

         private IntWritable line=new IntWritable();

         public void map(LongWritable key,Text value,Context context)throws IOException,InterruptedException{

             line.set(Integer.parseInt(value.toString()));

             context.write(line, new IntWritable(1));

         }

     }

     public static class Reduce extends Reducer<IntWritable,IntWritable,IntWritable,IntWritable>{

         private IntWritable num=new IntWritable(1);

         public void reduce(IntWritable key,Iterable<IntWritable> values,Context context)throws IOException,InterruptedException{

             for(IntWritable var:values){

             context.write(num, key);

             num=new IntWritable(num.get()+1);

             }

         }

     }

     public static class Partition extends Partitioner<IntWritable ,IntWritable>{

         @Override

         public int getPartition(IntWritable key, IntWritable value, int numPartitions) {

             // TODO Auto-generated method stub

             System.out.println(numPartitions);

             int maxnum=65223;

             int bound=maxnum/numPartitions+1;

             for(int i=0;i<numPartitions;i++)

             {

                 if(key.get()>=bound*(i-1)&&key.get()<=bound*i)

                 {

                     return i;

                 }

             }

             return 0;

         }

     }

     public int run(String[] args)throws Exception{

         Configuration conf=new Configuration();

         Job job=new Job(conf,"Sort");

         job.setJarByClass(Sort.class);

         job.setOutputKeyClass(IntWritable.class);

         job.setOutputValueClass(IntWritable.class);

         job.setMapperClass(Map.class);

         job.setReducerClass(Reduce.class);

         job.setPartitionerClass(Partition.class);

         job.setInputFormatClass(TextInputFormat.class);

         job.setOutputFormatClass(TextOutputFormat.class);

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

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

         boolean success=job.waitForCompletion(true);

         return success?0:1;

     }

     public static void main(String[] args)throws Exception{

         int ret=ToolRunner.run(new Sort(), args);

         System.exit(ret);

     }

 }