hadoop 中的mapreduce编程模板

public class MyClass extends Configured implements Tool {

     public static class MapClass extends MapReduceBase implements Mapper<Text,Text,Text,Text>{

         public void map(Text key,Text value,OutputCollector <Text,Text> output,Reporter reporter) throws IOException{

                output.collect(value,key);

   }

}

   

     public static class Reduce extends MapReduceBase implements Reducer <Text,Text,Text,Text> {

          public void reduce(Text key,Iterator<Text> values,OutputCollector<Text,Text>output,Report reporter) throws IOException {

        

            String csv="";

            while(value.hasNext()){

           

            if(csv.length()>0) csv+=",";

            csv+=values.next().toString();

            }

          output.collect(key,new Text(csv));

     }

 

}

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

   Configuration conf =getConf();

   

   JobConf job =new JobConf(conf ,MyJob.class);

   Path in=new Path(args[0]);

   Path out=new Path(args[1]);

   FileInputFormat.setInputPath(job,in);

   FileOutputFormat.setOutputPath(job,out);

  job.setJobName("MyJob");

  job.setMapperClass(MapClass.class);

  job.setReducerClass(Reduce.class);

  job.setInputFormat(KeyVlaueTextInputFormat.class);

  job.setOutputFormat(TextOutputFormat.class);

  job.setOutputKeyClass(Text.class);

   job.setOutputValueClass(Text.class);

  job.set("key.value.separator.in.input.line",",");

  JobClient.run.Job(job);

}

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

   int res = ToolRunner.run(new Configuration(),new MyClass(),args);

   System.exit(res);

}