Pulmonary_state

import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

   public class Pulomonary_states {

    public static class Map extends Mapper<LongWritable, Text, Text, Text> {

       private Text state = new Text();
       private Text disease = new Text();
       public void map(LongWritable key, Text value, Context context )
throws IOException, InterruptedException {
           String line = value.toString();
           String str[]=line.split(",");
           state.set(str[5]);
           disease.set(str[0]);
          if(state.toString().equals("AL") ||
state.toString().equalsIgnoreCase("il") ||         //to ignore the lower case  or upper case 
state.toString().equalsIgnoreCase("ak") ||
state.toString().equalsIgnoreCase("ca") ||
state.toString().equalsIgnoreCase("tx")){
               context.write(state, disease);
          }

       }
    }

    public static class Reduce extends Reducer<Text, Text, Text, IntWritable> {

       public void reduce(Text key, Iterable<Text> values, Context context)
         throws IOException, InterruptedException {
           int sum = 0;
           for (Text val : values) {
                   if(val.toString().contains("PULMONARY"))
               sum += 1;
           }
           
           context.write(key, new IntWritable(sum));
       }
    }

    public static void main(String[] args) throws Exception {
       Configuration conf = new Configuration();

           @SuppressWarnings("deprecation")
                Job job = new Job(conf, "wordcount");
           job.setJarByClass(Pulomonary_states.class);

           job.setMapOutputKeyClass(Text.class);  //in case output for mapper is different from..
           job.setMapOutputValueClass(Text.class); // ..the input of reducer 
       job.setOutputKeyClass(Text.class);
       job.setOutputValueClass(IntWritable.class);

       job.setMapperClass(Map.class);
       job.setReducerClass(Reduce.class);

       job.setInputFormatClass(TextInputFormat.class);
       job.setOutputFormatClass(TextOutputFormat.class);

       FileInputFormat.addInputPath(job, new Path(args[0]));
       FileOutputFormat.setOutputPath(job, new Path(args[1]));
        Path out=new Path(args[1]);
        out.getFileSystem(conf).delete(out);
       job.waitForCompletion(true);
    }

  }