import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class ParallelFrequentItemsetMiner extends Configured implements Tool
{
		
	public static void main(String args[]) throws Exception
	{
		if (args.length != 2)
		{
			System.out.println("usage: <path to input database> <path to output files>");
			System.exit(1); 
		}

		int res = ToolRunner.run(new ParallelFrequentItemsetMiner(), args);

		System.exit(res);
	}

	public int run(String args[]) throws Exception
	{
		long job_start_time, job_end_time; 
		long job_runtime; 
		
		JobConf conf = new JobConf(getConf()); 
			
		conf.setBoolean("mapred.reduce.tasks.speculative.execution", false); 
		conf.setInt("mapred.task.timeout", 60000000); 

		conf.setJarByClass(ParallelFrequentItemsetMiner.class);
			
		conf.setMapOutputKeyClass(IntWritable.class); 
		conf.setMapOutputValueClass(Text.class); 
			
		conf.setOutputKeyClass(Text.class); 
		conf.setOutputValueClass(DoubleWritable.class); 
			
		conf.setMapperClass(MapClass.class);
		//conf.setMapperClass(MapClassBinomialSampler.class);
		//conf.setMapperClass(MapClassCoinFlipSampler.class);
		conf.setReducerClass(ReduceClass.class);
			
		FileInputFormat.addInputPath(conf, new Path(args[0]));
		FileOutputFormat.setOutputPath(conf, new Path(args[1]));
			
		job_start_time = System.currentTimeMillis(); 
		JobClient.runJob(conf);
		job_end_time = System.currentTimeMillis(); 
			
		job_runtime = (job_end_time-job_start_time) / 1000; 
			
		System.out.println("total job runtime (seconds): " + job_runtime); 


		return 0;
	}
	
}


@@ -188,16 +188,68 @@
 o.Text;%0A
+import org.apache.hadoop.mapred.lib.IdentityMapper;%0A
 import o
@@ -423,16 +423,73 @@
 obConf;%0A
+import org.apache.hadoop.mapred.KeyValueTextInputFormat;%0A
 import o
@@ -726,9 +726,9 @@
  != 
-2
+3
 )%0A%09%09
@@ -804,12 +804,44 @@
 put 
-file
+local FIs%3E %3Cpath to output global FI
 s%3E%22)
@@ -2068,16 +2068,1028 @@
 ime); %0A%0A
+%09%09JobConf confAggr = new JobConf(getConf());%0A%0A%09%09confAggr.setBoolean(%22mapred.reduce.tasks.speculative.execution%22, false); %0A%09%09confAggr.setInt(%22mapred.task.timeout%22, 60000000); %0A%0A%09%09confAggr.setJarByClass(ParallelFrequentItemsetMiner.class);%0A%09%09%09%0A%09%09confAggr.setMapOutputKeyClass(Text.class); %0A%09%09confAggr.setMapOutputValueClass(Text.class); %0A%09%09%09%0A%09%09confAggr.setOutputKeyClass(Text.class); %0A%09%09confAggr.setOutputValueClass(DoubleWritable.class); %0A%09%09%09%0A%09%09confAggr.setMapperClass(IdentityMapper.class);%0A%09%09confAggr.setReducerClass(AggregateReducer.class);%0A%09%09%09%0A%09%09confAggr.setInputFormat(KeyValueTextInputFormat.class);%0A%0A%09%09KeyValueTextInputFormat.addInputPath(confAggr, new Path(args%5B1%5D));%0A%09%09FileOutputFormat.setOutputPath(confAggr, new Path(args%5B2%5D));%0A%0A%09%09job_start_time = System.currentTimeMillis(); %0A%09%09JobClient.runJob(confAggr);%0A%09%09job_end_time = System.currentTimeMillis(); %0A%09%09%09%0A%09%09job_runtime = (job_end_time-job_start_time) / 1000; %0A%09%09%09%0A%09%09System.out.println(%22total aggregation job runtime (seconds): %22 +%0A%09%09%09%09job_runtime); %0A
 %0A%09%09retur

Modify the driver class to execute the aggregate phase after we computed the local FI's from the sample. The number of input arguments increased from 2 to 3.