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");
    }

}