// Path: LUDecomposition/src/lud/naiveGaussian/mergeResults/merge_results_driver.java
// public class merge_results_driver {
// 
// 	public static boolean runWithJob(Job job, String out_path) throws IOException, InterruptedException, ClassNotFoundException {
// 	  job.setJarByClass(merge_results_driver.class);
// 	
// 	  job.setJobName("Final Step: Merging results and creating separate LU decomposed components of input matrix");
// 	
// 	  FileOutputFormat.setOutputPath(job, new Path(out_path));
// 	
// 	  job.setMapperClass(lud.naiveGaussian.mergeResults.merge_results_mapper.class);
// 	  job.setReducerClass(lud.naiveGaussian.mergeResults.merge_results_reducer.class);
// 	  job.setMapOutputKeyClass(TextPair.class);
// 	  job.setMapOutputValueClass(Text.class);
// 	  job.setOutputKeyClass(TextPair.class);
// 	  job.setOutputValueClass(Text.class);
// 	  job.setPartitionerClass(TextPairPartitioner.class);
//       job.setGroupingComparatorClass(NaturalKeyGroupingComparator.class);
//       job.setSortComparatorClass(TextPairComparator.class);
//       
//       boolean success = job.waitForCompletion(true);
// 	  return success;
// 	};
// }
// 
// Path: LUDecomposition/src/lud/naiveGaussian/totalRecords/total_records_driver.java
// public class total_records_driver {
// 	
// 	public static long readTotalRecords (String path, Configuration conf) throws IOException {
// 		FileSystem hdfs=FileSystem.get(conf);
// 		BufferedReader br = new BufferedReader(new InputStreamReader(hdfs.open(new Path(path+"/part-r-00000"))));
// 		Long records = (long) 0;
// 		records = Long.valueOf(br.readLine().split("\\t")[1]);
// 		br.close();
// 		hdfs.close();
// 		return records;
// 	}
// 
// 	@SuppressWarnings("deprecation")
// 	public static long run(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
// 	  Configuration conf = new Configuration();
// 	  Job job = new Job(conf);
// 	
// 	  job.setJarByClass(total_records_driver.class);
// 	
// 	  job.setJobName("Just counting total rows of the HDFS input");
// 	
// 	  FileInputFormat.setInputPaths(job, new Path(args[0]));
// 	  FileOutputFormat.setOutputPath(job, new Path(args[1]));
// 	
// 	  job.setMapperClass(total_records_mapper.class);
// 	
// 	  job.setReducerClass(total_records_reducer.class);
// 	  job.setCombinerClass(total_records_reducer.class);
// 	
// 	  job.setOutputKeyClass(LongWritable.class);
// 	  job.setOutputValueClass(LongWritable.class);
// 	  
// 	  //job.setInputFormatClass(TextInputFormat.class);
//       //job.setOutputFormatClass(TextOutputFormat.class);
// 	
// 	  job.waitForCompletion(true);
// 	  
// 	  return readTotalRecords(args[1], conf);
//   };
// }


import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.chain.ChainMapper;
import org.apache.hadoop.mapreduce.lib.chain.ChainReducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import lud.naiveGaussian.mergeResults.merge_results_driver;
import lud.naiveGaussian.totalRecords.total_records_driver;
			confLoop.set("mapreduce.job.reduce.slowstart.completedmaps", "1.00");
			confLoop.unset("n");
			confLoop.setLong("n", n);
			
			prepareJobWithConf(job, confLoop);
		}
		
		String lud_output_path = output+"/after-"+(total_records-1)+"-runs";
		job.setJarByClass(lud_driver.class);
		job.setJobName("Split a matrix into it's LU decomposed components using the Naive Gaussian Elimination method");
		FileInputFormat.addInputPath(job, new Path(input));
		FileOutputFormat.setOutputPath(job, new Path(lud_output_path));
		job.setInputFormatClass(KeyValueTextInputFormat.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		job.waitForCompletion(true);
		
		// MR Job(s): Merging Outputs
		
		Path merge_results_input_path = new Path(lud_output_path);
		conf.setBoolean("upper", false);
		job = new Job(conf);
		FileInputFormat.addInputPath(job, merge_results_input_path);
		String l_output_path = output+"/LU_Components/L";
		merge_results_driver.runWithJob(job, l_output_path);
		
		conf.setBoolean("upper", true);
		job = new Job(conf);
		String u_output_path = output+"/LU_Components/U";
		FileInputFormat.addInputPath(job, merge_results_input_path);
// Path: LUDecomposition/src/lud/naiveGaussian/mergeResults/merge_results_driver.java
// public class merge_results_driver {
// 
// 	public static boolean runWithJob(Job job, String out_path) throws IOException, InterruptedException, ClassNotFoundException {
// 	  job.setJarByClass(merge_results_driver.class);
// 	
// 	  job.setJobName("Final Step: Merging results and creating separate LU decomposed components of input matrix");
// 	
// 	  FileOutputFormat.setOutputPath(job, new Path(out_path));
// 	
// 	  job.setMapperClass(lud.naiveGaussian.mergeResults.merge_results_mapper.class);
// 	  job.setReducerClass(lud.naiveGaussian.mergeResults.merge_results_reducer.class);
// 	  job.setMapOutputKeyClass(TextPair.class);
// 	  job.setMapOutputValueClass(Text.class);
// 	  job.setOutputKeyClass(TextPair.class);
// 	  job.setOutputValueClass(Text.class);
// 	  job.setPartitionerClass(TextPairPartitioner.class);
//       job.setGroupingComparatorClass(NaturalKeyGroupingComparator.class);
//       job.setSortComparatorClass(TextPairComparator.class);
//       
//       boolean success = job.waitForCompletion(true);
// 	  return success;
// 	};
// }
// 
// Path: LUDecomposition/src/lud/naiveGaussian/totalRecords/total_records_driver.java
// public class total_records_driver {
// 	
// 	public static long readTotalRecords (String path, Configuration conf) throws IOException {
// 		FileSystem hdfs=FileSystem.get(conf);
// 		BufferedReader br = new BufferedReader(new InputStreamReader(hdfs.open(new Path(path+"/part-r-00000"))));
// 		Long records = (long) 0;
// 		records = Long.valueOf(br.readLine().split("\\t")[1]);
// 		br.close();
// 		hdfs.close();
// 		return records;
// 	}
// 
// 	@SuppressWarnings("deprecation")
// 	public static long run(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
// 	  Configuration conf = new Configuration();
// 	  Job job = new Job(conf);
// 	
// 	  job.setJarByClass(total_records_driver.class);
// 	
// 	  job.setJobName("Just counting total rows of the HDFS input");
// 	
// 	  FileInputFormat.setInputPaths(job, new Path(args[0]));
// 	  FileOutputFormat.setOutputPath(job, new Path(args[1]));
// 	
// 	  job.setMapperClass(total_records_mapper.class);
// 	
// 	  job.setReducerClass(total_records_reducer.class);
// 	  job.setCombinerClass(total_records_reducer.class);
// 	
// 	  job.setOutputKeyClass(LongWritable.class);
// 	  job.setOutputValueClass(LongWritable.class);
// 	  
// 	  //job.setInputFormatClass(TextInputFormat.class);
//       //job.setOutputFormatClass(TextOutputFormat.class);
// 	
// 	  job.waitForCompletion(true);
// 	  
// 	  return readTotalRecords(args[1], conf);
//   };
// }


// Path: LUDecomposition/src/lud/naiveGaussian/lud_driver.java
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.chain.ChainMapper;
import org.apache.hadoop.mapreduce.lib.chain.ChainReducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import lud.naiveGaussian.mergeResults.merge_results_driver;
import lud.naiveGaussian.totalRecords.total_records_driver;

			confLoop.set("mapreduce.job.reduce.slowstart.completedmaps", "1.00");
			confLoop.unset("n");
			confLoop.setLong("n", n);
			
			prepareJobWithConf(job, confLoop);
		}
		
		String lud_output_path = output+"/after-"+(total_records-1)+"-runs";
		job.setJarByClass(lud_driver.class);
		job.setJobName("Split a matrix into it's LU decomposed components using the Naive Gaussian Elimination method");
		FileInputFormat.addInputPath(job, new Path(input));
		FileOutputFormat.setOutputPath(job, new Path(lud_output_path));
		job.setInputFormatClass(KeyValueTextInputFormat.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		job.waitForCompletion(true);
		
		// MR Job(s): Merging Outputs
		
		Path merge_results_input_path = new Path(lud_output_path);
		conf.setBoolean("upper", false);
		job = new Job(conf);
		FileInputFormat.addInputPath(job, merge_results_input_path);
		String l_output_path = output+"/LU_Components/L";
		merge_results_driver.runWithJob(job, l_output_path);
		
		conf.setBoolean("upper", true);
		job = new Job(conf);
		String u_output_path = output+"/LU_Components/U";
		FileInputFormat.addInputPath(job, merge_results_input_path);
		merge_results_driver.runWithJob(job, u_output_path);