package org.apache.hadoop.examples;

 import java.io.IOException;

 import java.util.Iterator;

 import java.util.StringTokenizer;

 import org.apache.hadoop.conf.Configuration;

 import org.apache.hadoop.fs.Path;

 import org.apache.hadoop.io.IntWritable;

 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;

 public class sort {

     public sort() {

     }

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

         Configuration conf = new Configuration();

         String fileAddress = "hdfs://localhost:9000/user/hadoop/";

         //String[] otherArgs = (new GenericOptionsParser(conf, args)).getRemainingArgs();

         String[] otherArgs = new String[]{fileAddress+"number.txt", fileAddress+"output"};

         if(otherArgs.length < 2) {

             System.err.println("Usage: sort <in> [<in>...] <out>");

             System.exit(2);

         }

         Job job = Job.getInstance(conf, "sort");

         job.setJarByClass(sort.class);

         job.setMapperClass(sort.TokenizerMapper.class);

         //job.setCombinerClass(sort.SortReducer.class);

         job.setReducerClass(sort.SortReducer.class);

         job.setOutputKeyClass(IntWritable.class);

         job.setOutputValueClass(IntWritable.class);

         for(int i = 0; i < otherArgs.length - 1; ++i) {

             FileInputFormat.addInputPath(job, new Path(otherArgs[i]));

         }

         FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1]));

         System.exit(job.waitForCompletion(true)?0:1);

     }

     public static class TokenizerMapper extends Mapper<Object, Text, IntWritable, IntWritable> {

         public TokenizerMapper() {

         }

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

             StringTokenizer itr = new StringTokenizer(value.toString());

             while(itr.hasMoreTokens()) {

                 context.write(new IntWritable(Integer.parseInt(itr.nextToken())), new IntWritable(1));

             }

         }

     }

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

         private static IntWritable num = new IntWritable(1);

         public SortReducer() {

         }

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

             for(Iterator<IntWritable> i$ = values.iterator(); i$.hasNext();i$.next()) {

                 context.write(num, key);

             }

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

         }

     }

 }