import java.io.IOException;
import java.util.Arrays;
import java.util.ArrayList;
import java.util.Iterator;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
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 AggregateReducer extends MapReduceBase 
	implements Reducer<Text, DoubleWritable, Text, Text>
{
	int reducersNum;
	int reqApproxNum;
	int sampleSize;
	float epsilon;

	@Override
	public void configure(JobConf conf) 
	{
		reducersNum = conf.getInt("PARMM.reducersNum", 64);
		reqApproxNum = conf.getInt("PARMM.reqApproxNum", reducersNum / 2 +1);
		epsilon = conf.getFloat("PARMM.epsilon", (float) 0.02);
		sampleSize = conf.getInt("PARMM.sampleSize", 1000);
	}



	@Override
	public void reduce(Text itemset, Iterator<DoubleWritable> values, 
			OutputCollector<Text,Text> output, 
			Reporter reporter) throws IOException
	{
		ArrayList<Double> valuesArrList = new ArrayList<Double>();
		while (values.hasNext()) 
		{
			String valueString = (values.next()).toString();
			valuesArrList.add(Double.parseDouble(valueString));
		}
		/**
		 * Only consider the itemset as "global frequent" if it
		 * appears among the "local frequent" itemsets a sufficient
		 * number of times.
		 */
		if (valuesArrList.size() >= reqApproxNum)
		{
			Double[] valuesArr = new Double[valuesArrList.size()];
			valuesArrList.toArray(valuesArr);
			Arrays.sort(valuesArr);

			/**
			 * Compute the smallest frequency interval containing
			 * requiredNum estimates of the frequency of the
			 * itemset. Use the center of this interval as global
			 * estimate for the frequency. The confidence interval
			 * is obtained by enlarging the above interval by
			 * epsilon/2 on both sides.
			 */
			int intervalPoints = reducersNum - reqApproxNum + 1;
			double minIntervalLength = valuesArr[intervalPoints- 1] - valuesArr[0];
			double estimatedFreq = (valuesArr[0] + minIntervalLength / 2);
			for (int i = 1; i < valuesArr.length - intervalPoints; i++)
			{
				double intervalLength = valuesArr[intervalPoints + i] - valuesArr[i];
				if (intervalLength < minIntervalLength) 
				{
					minIntervalLength = intervalLength;
					estimatedFreq = valuesArr[i] + minIntervalLength / 2;
				}
			}
			double confIntervalLowBound = (estimatedFreq - minIntervalLength / 2) / sampleSize - epsilon / 2;
			double confIntervalUppBound = (estimatedFreq + minIntervalLength / 2) / sampleSize + epsilon / 2;
			estimatedFreq = estimatedFreq / sampleSize;
			
			String estFreqAndBoundsStr = "(" + estimatedFreq + "," + confIntervalLowBound + "," + confIntervalUppBound + ")"; 
			output.collect(itemset, new Text(estFreqAndBoundsStr));
		} // end if (valuesArrList.size() >= requiredNum)
	}
}



@@ -1931,26 +1931,388 @@
 %0A%09%09%09
-int intervalPoints
+double minIntervalLength = valuesArr%5BreqApproxNum - 1%5D - valuesArr%5B0%5D;%0A%09%09%09int startIndex = 0;%0A%09%09%09for (int i = 1; i %3C valuesArr.length - reqApproxNum; i++)%0A%09%09%09%7B%0A%09%09%09%09double intervalLength = valuesArr%5BreqApproxNum - 1 + i%5D - valuesArr%5Bi%5D;%0A%09%09%09%09if (intervalLength %3C minIntervalLength) %0A%09%09%09%09%7B%0A%09%09%09%09%09minIntervalLength = intervalLength;%0A%09%09%09%09%09startIndex = i;%0A%09%09%09%09%7D%0A%09%09%09%7D%0A%09%09%09int reqApproxNum2
  = r
@@ -2341,31 +2341,124 @@
 Num + 1;%0A%09%09%09
-d
+D
 ouble
- 
+%5B%5D valuesSubArr = Arrays.copyOfRange(valuesArr, startIndex, startIndex + reqApproxNum -1);%0A%09%09%09
 minIntervalL
@@ -2471,34 +2471,37 @@
 = values
+Sub
 Arr%5B
-intervalPoints
+reqApproxNum2 
 - 1%5D - v
@@ -2497,32 +2497,35 @@
 m2 - 1%5D - values
+Sub
 Arr%5B0%5D;%0A%09%09%09doubl
@@ -2545,24 +2545,27 @@
 eq = (values
+Sub
 Arr%5B0%5D + min
@@ -2582,14 +2582,15 @@
 ngth
+)
  / 2
-)
 ;
+ 
 %0A%09%09%09
@@ -2607,32 +2607,35 @@
  = 1; i %3C values
+Sub
 Arr.length - int
@@ -2631,30 +2631,29 @@
 ength - 
-intervalPoints
+reqApproxNum2
 ; i++)%0A%09
@@ -2694,26 +2694,32 @@
 lues
+Sub
 Arr%5B
-intervalPoints
+reqApproxNum2 - 1
  + i
@@ -2720,32 +2720,35 @@
  1 + i%5D - values
+Sub
 Arr%5Bi%5D;%0A%09%09%09%09if (
@@ -2854,24 +2854,27 @@
 req = values
+Sub
 Arr%5Bi%5D + min
@@ -2888,24 +2888,25 @@
 lLength / 2;
+ 
 %0A%09%09%09%09%7D%0A%09%09%09%7D%0A
@@ -2905,16 +2905,17 @@
 %09%7D%0A%09%09%09%7D%0A
+%0A
 %09%09%09doubl

Improve computation of the confidence interval.