package de.hsma.bdea;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.log4j.BasicConfigurator;

// Formel und Bsp. von http://www.crashkurs-statistik.de/der-korrelationskoeffizient-nach-pearson

public class Korrelation {

	public static void main(String[] args) throws Exception {
		BasicConfigurator.configure(); 					// Log4j Config oder ConfigFile in Resources Folder
		System.setProperty("hadoop.home.dir", "/");  	// für Hadoop 3.3.0
		
		args = new String[] {"resources/corr.txt", "/tmp/output-avg/", "/tmp/output-sq/", "/tmp/output-sumRes/", "/tmp/output-corr/"};

		// Job 1
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf, "average");
		job.setNumReduceTasks(2);

		job.setJarByClass(Korrelation.class);
		job.setMapperClass(TokenizerMapper.class);
		job.setReducerClass(AverageReducer.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(DoubleWritable.class);
		job.setOutputFormatClass(SequenceFileOutputFormat.class);

		FileInputFormat.addInputPath(job, new Path(args[0]));

		Path outputPath = new Path(args[1]);
		FileOutputFormat.setOutputPath(job, outputPath);
		outputPath.getFileSystem(conf).delete(outputPath, true);

		job.waitForCompletion(true);

		// -----------------------
		// Job 2

		conf = new Configuration();
		job = Job.getInstance(conf, "squaring");
		job.setNumReduceTasks(2);

		job.setJarByClass(Korrelation.class);
		job.setMapperClass(SquareMapper.class);
		job.setReducerClass(Reducer.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(DoubleWritable.class);

		FileInputFormat.addInputPath(job, new Path(args[0]));
		job.setOutputFormatClass(SequenceFileOutputFormat.class);

		outputPath = new Path(args[2]);
		FileOutputFormat.setOutputPath(job, outputPath);
		outputPath.getFileSystem(conf).delete(outputPath, true);

		job.waitForCompletion(true);

		// -----------------------
		// Job 3
		
		conf = new Configuration();
		job = Job.getInstance(conf, "sumResults");
		job.setNumReduceTasks(2);

		job.setJarByClass(Korrelation.class);

		MultipleInputs.addInputPath(job, new Path(args[1]), SequenceFileInputFormat.class, Mapper.class);
		MultipleInputs.addInputPath(job, new Path(args[2]), SequenceFileInputFormat.class, Mapper.class);

		job.setReducerClass(DoubleSumReducer.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(DoubleWritable.class);
		job.setOutputFormatClass(SequenceFileOutputFormat.class);

		outputPath = new Path(args[3]);
		FileOutputFormat.setOutputPath(job, outputPath);
		outputPath.getFileSystem(conf).delete(outputPath, true);

		job.waitForCompletion(true);

		// -----------------------
		// Job 4

		conf = new Configuration();
		job = Job.getInstance(conf, "finalize");
		job.setNumReduceTasks(1);

		job.setJarByClass(Korrelation.class);
		job.setMapperClass(Mapper.class);
		job.setReducerClass(FinalReducer.class);

		job.setInputFormatClass(SequenceFileInputFormat.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(DoubleWritable.class);

		FileInputFormat.addInputPath(job, new Path(args[3]));
		outputPath = new Path(args[4]);
		FileOutputFormat.setOutputPath(job, outputPath);
		outputPath.getFileSystem(conf).delete(outputPath, true);

		job.waitForCompletion(true);		
		System.exit(0);
	}
	
	// -------------------------------
	// Job 1
	
	public static class TokenizerMapper extends Mapper<LongWritable, Text, Text, DoubleWritable> {
		private final static DoubleWritable dw = new DoubleWritable();
		private Text[] elem = {new Text("xvg"), new Text("yvg")};

		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String[] spalten = value.toString().split(",");

			for (int i = 1; i < 3; i++) {
				dw.set(Double.parseDouble(spalten[i]));
				context.write(elem[i - 1], dw);
			}
		}
	}
	
	public static class AverageReducer extends Reducer<Text, DoubleWritable, Text, DoubleWritable> {
		private DoubleWritable result = new DoubleWritable();

		public void reduce(Text key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {
			double sum = 0;
			long cnt = 0;

			for (DoubleWritable val : values) {
				sum += val.get();
				cnt++;
			}
			
			result.set(sum/cnt);
			context.write(key, result);
		}
	}

	// ---------
	// Job 2
	
	public static class SquareMapper extends Mapper<LongWritable, Text, Text, DoubleWritable> {
		private final static DoubleWritable dw = new DoubleWritable();

		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			double xsq = 0;
			double ysq = 0;
			double sum = 0;
			int cnt = 1;

			String[] spalten = value.toString().split(",");

			double x = Double.parseDouble(spalten[1]);
			double y = Double.parseDouble(spalten[2]);
			
			xsq += x*x;
			ysq += y*y;
			sum += x*y;

			dw.set(xsq);
			context.write(new Text("xsq"), dw);
			dw.set(ysq);
			context.write(new Text("ysq"), dw);
			dw.set(sum);
			context.write(new Text("sum"), dw);
			dw.set(cnt);
			context.write(new Text("cnt"), dw);
		}
	}
	
	// ---------
	// Job 3
	
	public static class DoubleSumReducer extends Reducer<Text, DoubleWritable, Text, DoubleWritable> {
		private DoubleWritable result = new DoubleWritable();

		public void reduce(Text key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {
			double sum = 0;
			for (DoubleWritable val : values) {
				sum += val.get();
			}
			result.set(sum);
			context.write(key, result);
		}
		
	}

	// --------
	// Job 4
	
	public static class FinalReducer extends Reducer<Text, DoubleWritable, Text, DoubleWritable> {
		private DoubleWritable result = new DoubleWritable();
		private Map<String, Double> map = new HashMap<>();
	
		public void reduce(Text key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {
			for (DoubleWritable val : values) {
				map.put(key.toString(), val.get());
			}
		}
		
		public void cleanup(Context context) throws IOException, InterruptedException {
			long cnt = (long) (double) map.get("cnt");
			
			double r = (map.get("sum") - (cnt * map.get("xvg") * map.get("yvg"))) /
					   (Math.sqrt(map.get("xsq") - cnt * map.get("xvg") * map.get("xvg")) 
					  * Math.sqrt(map.get("ysq") - cnt * map.get("yvg") * map.get("yvg")));
			result.set(r);

			context.write(new Text("correlation"), result);
		}
	}

}




