package query;

import java.io.IOException;
import java.util.Iterator;
import java.util.HashMap;
import java.util.Set;

import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

public class RrdReducer extends MapReduceBase implements Reducer<IntWritable, FloatWritable, IntWritable, FloatWritable> {

	public void reduce(IntWritable key, Iterator<FloatWritable> values, OutputCollector<IntWritable, FloatWritable> output, Reporter reporter) throws IOException
	{
		float threshold = 90;
		int time = 120;
		int min;
		int max;
		int cpt=0;
		HashMap<Integer, Float> cons = new HashMap<Integer, Float>();
		while (values.hasNext())
		{
			System.out.println("cpt: "+cpt++);
			Float v = values.next().get();
			if(v > threshold)
			{
				System.out.println("key: "+ key.get()+" value: "+v);
				cons.put(key.get(), v);
			}
			else {
				if(cons.size()>0) {
					Set<Integer> key_set = cons.keySet();
					min = Integer.MAX_VALUE;
					max = 0;
					for(Integer i : key_set)
					{
						if(i>max)
							max = i;
						if(i<min)
							min=i;
					}
					System.out.println("MIN: "+min);
					System.out.println("MAX: "+max);

					if((max - min) > time)
						for(Integer k : key_set)
						{
							v = cons.get(k);
							output.collect(new IntWritable(k), new FloatWritable(v));
						}
					cons.clear();
				}
			}
		}
	}

}