package jesg;

import java.io.IOException;

import jesg.avro.Pair;

import org.apache.avro.generic.GenericRecord;
import org.apache.avro.mapred.AvroKey;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;

class FreqReducer extends Reducer<
	AvroKey<Pair>, IntWritable, 
	AvroKey<Pair>, DoubleWritable> {
	
	private CharSequence currentWord;
	private int currentTotal = 0;
	
	@Override
	public void reduce(AvroKey<Pair> key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
		GenericRecord record = key.datum();
		
		if( (Boolean)record.get("special") ) {
			currentTotal = sum(values);
			currentWord = (CharSequence) record.get("first");
			return;
		}
		
		if( !((CharSequence) record.get("first")).equals(currentWord) ){
			System.out.println(">>>>>>>>>>>>>>>>>>>>ERRRRRRROR");
		}
		
		context.write(key, new DoubleWritable((double)sum(values)/currentTotal));
		
	}
	
	private int sum(Iterable<IntWritable> values){
		int sum = 0;
		for (IntWritable i : values) {
			sum += i.get();
		}
		return sum;
	}
}

@@ -379,44 +379,8 @@
 %3E %7B%0A
-%09%0A%09private CharSequence currentWord;
 %0A%09pr
@@ -401,20 +401,16 @@
 entTotal
- = 0
 ;%0A%09%0A%09@Ov
@@ -665,198 +665,14 @@
 %0A%09%09%09
-currentWord = (CharSequence) record.get(%22first%22);%0A%09%09%09return;%0A%09%09%7D%0A%09%09%0A%09%09if( !((CharSequence) record.get(%22first%22)).equals(currentWord) )%7B%0A%09%09%09System.out.println(%22%3E%3E%3E%3E%3E%3E%3E%3E%3E%3E%3E%3E%3E%3E%3E%3E%3E%3E%3E%3EERRRRRRROR%22)
+return
 ;%0A%09%09

remove syso