source: proiecte/PDAD/trunk/endreason/src/ro/pub/cs/pp/pdad/EndReason.java @ 154

package ro.pub.cs.pp.pdad;

import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.IntWritable;
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 EndReason {

        private static int[] domains = {-1, 0, 1, 1000, 2000, 3000, 4000, 5000, 6000, 7000};
        private static String[] reasonName = {"not reported", "undetermined", "infrastructure", "hardware", "IO", "network", "software", "human error", "user"}; 
        
        public static class Map extends Mapper<Object, Text, IntWritable, IntWritable> {
                
                private static String null_reason = "NULL"; 
                
                private final static IntWritable one = new IntWritable(1); 
                private final static IntWritable zero = new IntWritable(0); 
                private final static int fieldNo = 9; 
                
                private int findDomainCode(int reasonCode) {
                        for (int domainCode = 0; domainCode < domains.length - 1; domainCode++) {
                                if (domains[domainCode] <= reasonCode && reasonCode < domains[domainCode + 1])
                                        return domainCode; 
                        }
                        return 0; 
                }
                
                private int extractReasonDomainCode(String line) {
                        String[] pieces = line.split("\\s+");
                        if (pieces.length < fieldNo + 1) {
                                return 0; 
                        }
                        String reason = pieces[fieldNo]; 
                        if (reason.equals(null_reason)) {
                                return 0;                               
                        } else {
                                int reasonCode = Integer.parseInt(reason);
                                return findDomainCode(reasonCode); 
                        }
                }
                
                public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
                        String line = value.toString(); 
                        try { 
                                int reasonDomain = extractReasonDomainCode(line);
                                context.write(new IntWritable(reasonDomain), one);
                        } catch (NumberFormatException ex) {
                                context.write(zero, zero); 
                        }
                        
                }
        }
        
        public static class Reduce extends Reducer <IntWritable, IntWritable, Text, IntWritable> {
                
                private Text reasonDomain = new Text(); 
                
                public void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
                        int sum = 0; 
                        for (IntWritable cnt : values) {
                                sum += cnt.get(); 
                        }
                        reasonDomain.set(reasonName[key.get()]); 
                        context.write(reasonDomain, new IntWritable(sum));  
                }
        }
        
        public static void main(String[] args) throws Exception {
                Configuration conf = new Configuration(); 
                Job job = new Job(conf, "EndReason"); 
                job.setNumReduceTasks(5);
                job.setJarByClass(EndReason.class);
                
                /* set map output key/value classes */
                job.setMapOutputKeyClass(IntWritable.class); 
                job.setMapOutputValueClass(IntWritable.class); 
                
                /* set application output key/value classes */
                job.setOutputKeyClass(Text.class); 
                job.setOutputValueClass(IntWritable.class);
                
                job.setMapperClass(Map.class);
                job.setReducerClass(Reduce.class);              
                /* don't set any combiner class */
                
                FileInputFormat.addInputPath(job, new Path(args[0])); 
                FileOutputFormat.setOutputPath(job, new Path(args[1])); 
                System.exit(job.waitForCompletion(true) ? 0 : 1);
        }
        
}