// cc OldMaxTemperature Aplikacja określająca temperaturę maksymalną, używająca 
// dawnego interfejsu API dla modelu MapReduce
import java.io.IOException;
import java.util.Iterator;

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

// vv OldMaxTemperature
public class OldMaxTemperature {
  
  static class OldMaxTemperatureMapper /*[*/extends MapReduceBase/*]*/
    /*[*/implements Mapper/*]*/<LongWritable, Text, Text, IntWritable> {
  
    private static final int MISSING = 9999;
    
    @Override
    public void map(LongWritable key, Text value,
        /*[*/OutputCollector<Text, IntWritable> output, Reporter reporter/*]*/)
        throws IOException {
      
      String line = value.toString();
      String year = line.substring(15, 19);
      int airTemperature;
      if (line.charAt(87) == '+') { // Metoda parseInt źle radzi sobie ze znakami plus na początku
        airTemperature = Integer.parseInt(line.substring(88, 92));
      } else {
        airTemperature = Integer.parseInt(line.substring(87, 92));
      }
      String quality = line.substring(92, 93);
      if (airTemperature != MISSING && quality.matches("[01459]")) {
        /*[*/output.collect/*]*/(new Text(year), new IntWritable(airTemperature));
      }
    }
  }
  
  static class OldMaxTemperatureReducer /*[*/extends MapReduceBase/*]*/
    /*[*/implements Reducer/*]*/<Text, IntWritable, Text, IntWritable> {

    @Override
    public void reduce(Text key, /*[*/Iterator/*]*/<IntWritable> values,
        /*[*/OutputCollector<Text, IntWritable> output, Reporter reporter/*]*/)
        throws IOException {
      
      int maxValue = Integer.MIN_VALUE;
      while (/*[*/values.hasNext()/*]*/) {
        maxValue = Math.max(maxValue, /*[*/values.next().get()/*]*/);
      }
      /*[*/output.collect/*]*/(key, new IntWritable(maxValue));
    }
  }

  public static void main(String[] args) throws IOException {
    if (args.length != 2) {
      System.err.println("Użytkowanie: OldMaxTemperature <ścieżka wej.> <ścieżka wyj.>");
      System.exit(-1);
    }
    
    /*[*/JobConf conf = new JobConf(OldMaxTemperature.class);/*]*/
    /*[*/conf/*]*/.setJobName("Max temperature");

    FileInputFormat.addInputPath(/*[*/conf/*]*/, new Path(args[0]));
    FileOutputFormat.setOutputPath(/*[*/conf/*]*/, new Path(args[1]));
    
    /*[*/conf/*]*/.setMapperClass(OldMaxTemperatureMapper.class);
    /*[*/conf/*]*/.setReducerClass(OldMaxTemperatureReducer.class);

    /*[*/conf/*]*/.setOutputKeyClass(Text.class);
    /*[*/conf/*]*/.setOutputValueClass(IntWritable.class);

    /*[*/JobClient.runJob(conf);/*]*/
  }
}
// ^^ OldMaxTemperature