mapreduce代码实现入门

　　mapreduce代码主要包括三个类，map类、reduce类以及测试类！

以wordcount为例，

map类为：

    static class WordMapper extends Mapper<Object, Text, Text, IntWritable>{

        private final static IntWritable one = new IntWritable(1);

        private Text word = new Text();

        public void map(Object key, Text value, Context context)

            throws IOException, InterruptedException{

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

            while (itr.hasMoreElements()) {

                word.set(itr.nextToken());

                context.write(word, one);

            }

        }

    }

reduce类为：

    static class WordReducer extends Reducer<Text, IntWritable, Text, IntWritable>{

        private IntWritable res = new IntWritable();

        public void reduce(Text key, Iterable<IntWritable> values, Context context)

            throws IOException, InterruptedException

        {

            int sum = 0;

            for(IntWritable val:values){

                sum += val.get();

            }

            res.set(sum);

            context.write(key, res);

        }

    }

主函数代码为：

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

        String inputfilepath = "hdfs://localhost:9000/input1";

        String outputfilepath = "hdfs://localhost:9000/output4";

        Configuration conf = new Configuration();

        Job job = new Job(conf);

        job.setJarByClass(WordCount.class);

        job.setJobName("word-count");

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(IntWritable.class);

        job.setMapperClass(WordMapper.class);

        job.setReducerClass(WordReducer.class);

        job.setInputFormatClass(TextInputFormat.class);

        job.setOutputFormatClass(TextOutputFormat.class);

        FileInputFormat.addInputPath(job, new Path(inputfilepath));

        FileOutputFormat.setOutputPath(job, new Path(outputfilepath));

        job.waitForCompletion(true);

    }

其他的hadoop简单实例代码如：

数字求和：

 package goal;

 import java.io.IOException;

 import java.util.StringTokenizer;

 import org.apache.hadoop.conf.Configuration;

 import org.apache.hadoop.fs.Path;

 import org.apache.hadoop.io.FloatWritable;

 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.apache.hadoop.util.GenericOptionsParser;

 public class Sum {

     public static class SumMapper extends

         Mapper<Object, Text, Text, FloatWritable>{

         private Text word = new Text("sum");

         private static FloatWritable nv = new FloatWritable(1.0f);

         public void map(Object key, Text value, Context context)

             throws IOException, InterruptedException

         {

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

             float sum = 0;

             while(str.hasMoreTokens()){

                 String s = str.nextToken();

                 float val = Float.parseFloat(s);

                 sum = val;

             }

             nv.set(sum);

             context.write(word, nv);

         }

     }

     public static class SumReducer extends

         Reducer<Text, FloatWritable, Text, FloatWritable>{

         private Text k = new Text("sum");

         private FloatWritable res = new FloatWritable();

         public void reduce(Text key, Iterable<FloatWritable> values,

                 Context context) throws IOException, InterruptedException{

             float sum = 0;

             for(FloatWritable val : values){

                 float v = val.get();

                 sum += v;

             }

             res.set(sum);

             context.write(k, res);

         }

     }

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

         String other[] = {"hdfs://localhost:9000/input2/1.txt", "hdfs://localhost:9000/output3"};

         Configuration conf = new Configuration();

         System.out.println("yes");

         Job job = new Job(conf, "number sum");

         job.setJarByClass(Sum.class);

         job.setMapperClass(SumMapper.class);

         job.setReducerClass(SumReducer.class);

         job.setOutputKeyClass(Text.class);

         job.setOutputValueClass(FloatWritable.class);

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

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

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

         System.out.println("yes");

     }

 }

巴特西

mapreduce代码实现入门

最新文章

热门文章