source: proiecte/PDAD/trunk/mediumevents/mapreduce/MediumEvents.java @ 154

/**
 * Medium length for events
 * 
 * @author cristina
 */

package org.PP;

import java.io.IOException;
import java.util.*;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;

public class MediumEvents {

        public static class Map extends MapReduceBase implements
                        Mapper<LongWritable, Text, Text, DoubleWritable> {
                private static Text label = new Text();
                static long anotherLabel = 0; // groups data by generating a new lable
                                                                                // each 1000 entries
                static long labels = 0; // number of keys generated

                public enum MapDuration {
                        AHAM
                };

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

                        Map.label.set("medium duration " + Map.labels);

                        if (Map.anotherLabel == 1000) {
                                Map.anotherLabel = 0;
                                Map.labels++;

                                // reporter.setStatus("map " + label.toString());
                                reporter.incrCounter(MapDuration.AHAM, 1);
                        }

                        String line = value.toString();
                        // skip comments
                        if (!line.startsWith("#")) {
                                // collons 7 and 8 are start, end time of job respectively
                                StringTokenizer tokenizer = new StringTokenizer(line);
                                int count = 0;
                                Double start = new Double(0), end = new Double(0), duration = new Double(
                                                0);
                                anotherLabel++;
                                while (tokenizer.hasMoreTokens()) {
                                        count++;
                                        String token = tokenizer.nextToken();
                                        if (count == 7)
                                                start = Double.parseDouble(token);
                                        if (count == 8) {
                                                end = Double.parseDouble(token);
                                                // stop, we have all the information
                                                break;
                                        }
                                }

                                duration = end - start;
                                output.collect(label, new DoubleWritable(duration));
                        }
                }
        }

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

                public void reduce(Text key, Iterator<DoubleWritable> values,
                                OutputCollector<Text, DoubleWritable> output, Reporter reporter)
                                throws IOException {
                        Double sum = new Double(0);
                        Long count = new Long(0);
                        Double med = new Double(0);

                        while (values.hasNext()) {
                                count++;
                                sum += values.next().get();
                        }
                        if (count > 0)
                                med = sum / count;
                        else
                                med = new Double(0);
                        output.collect(new Text("medium duration"), new DoubleWritable(med));

                }
        }

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

        public void reduce(Text key, Iterator<DoubleWritable> values,
                        OutputCollector<Text, DoubleWritable> output, Reporter reporter)
                        throws IOException {
                Double sum = new Double(0);
                Long count = new Long(0);
                Double med = new Double(0);
        
                while (values.hasNext()) {
                        count++;
                        sum += values.next().get();
                }
                if (count > 0)
                        med = sum / count;
                else
                        med = new Double(0);
                output.collect(key, new DoubleWritable(med));
        
        }
}
        
        /*
         * public static class Reduce extends MapReduceBase implements Reducer<Text,
         * DoubleWritable, Text, DoubleWritable> {
         * 
         * static Text label = new Text(); static long anotherLabel = 0; // groups
         * data by generating a new lable each 1.000.000 entries static long labels
         * = 0; // number of keys generated
         * 
         * public enum RedDuration { AHAM };
         * 
         * 
         * public void reduce(Text key, Iterator<DoubleWritable> values,
         * OutputCollector<Text, DoubleWritable> output, Reporter reporter) throws
         * IOException { Double sum = new Double(0); Long count = new Long(0);
         * Double med = new Double(0);
         * 
         * Reduce.label.set("medium duration "+Reduce.labels);
         * 
         * if(Reduce.anotherLabel == 100) { Reduce.anotherLabel = 0;
         * Reduce.labels++;
         * 
         * //reporter.setStatus("map " + label.toString());
         * reporter.incrCounter(RedDuration.AHAM, 1); }
         * 
         * Reduce.anotherLabel++;
         * 
         * 
         * while (values.hasNext()) { count++; sum += values.next().get(); }
         * if(count > 0) med = sum/count; else med = new Double(0);
         * output.collect(Reduce.label, new DoubleWritable(med));
         * 
         * } }
         */

        public static void main(String[] args) throws Exception {
                JobConf conf = new JobConf(MediumEvents.class);
                conf.setJobName("mediumEvents");

                conf.setOutputKeyClass(Text.class);
                conf.setOutputValueClass(DoubleWritable.class);

                conf.setMapperClass(Map.class);
                conf.setCombinerClass(Combiner.class);
                conf.setReducerClass(Reduce.class);

                conf.setInputFormat(TextInputFormat.class);
                conf.setOutputFormat(TextOutputFormat.class);

                FileInputFormat.setInputPaths(conf, new Path(args[0]));
                FileOutputFormat.setOutputPath(conf, new Path(args[1]));

                conf.setNumReduceTasks(5);

                JobClient.runJob(conf);
        }
}