package de.hsma.bdea;

import java.io.IOException;
import java.util.Iterator;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
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;

//10.0
//22.0
//50.0

public class MVM2 {
	static int length;

	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/mvm.txt", "resources/mvm-output-1" + System.currentTimeMillis(), "resources/mvm-output-2" + System.currentTimeMillis()};

		Configuration conf = new Configuration();
//		conf.set("length", "3");
		length = 3; // Anzahl Spalten in Matrix (== Zeilen Vector) (ggf. mit einem vorherigen Job ermitteln)
		Job job = Job.getInstance(conf, "mvm1");

		job.setJarByClass(MVM2.class);
		job.setMapperClass(TokenizerMapper.class);
		job.setReducerClass(MVMMultiplier.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.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();
//		conf.set("length", "3");
		job = Job.getInstance(conf, "mvm2");

		job.setJarByClass(MVM2.class);
		job.setMapperClass(Mapper.class);
		job.setReducerClass(MVMReducer.class);
		job.setNumReduceTasks(2);
		job.setPartitionerClass(VectorPartitioner.class); // zur Sortierung nach Vektor Koordinate (optional)
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);

		FileInputFormat.addInputPath(job, outputPath);
		job.setInputFormatClass(SequenceFileInputFormat.class);

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

		job.waitForCompletion(true);
		System.exit(0);
	}

	public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {

		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
			String line = value.toString();
			if ("".equals(line)) {
				return;
			}

			//			System.out.println("+++ " + context.getTaskAttemptID().getTaskID().getId() + " - " + value.toString());
//			int length = Integer.parseInt(context.getConfiguration().get("length"));

			String k = null;
			String v = null;

			StringTokenizer st = new StringTokenizer(line, ",");

			st.nextToken();	// ignore M or V

			if (line.startsWith("M")) {
				k = st.nextToken() + "," + st.nextToken();
				v = st.nextToken();
				context.write(new Text(k), new IntWritable(Integer.parseInt(v)));
				//					System.out.println(k + " " + v);
			} else {
				k = st.nextToken();
				v = st.nextToken();

				for (int i = 0; i < length; i++) {
					//						System.out.println(i+"," + k + " " + v);
					context.write(new Text(i + "," + k), new IntWritable(Integer.parseInt(v)));
				}
			}

		}
	}

	public static class MVMMultiplier extends Reducer<Text, IntWritable, Text, IntWritable> {
		private IntWritable result = new IntWritable();
		private Text newKey = new Text();

		public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
//			System.out.println("mmmmkkkkkk " + key.toString());

			Iterator<IntWritable> it = values.iterator();
			result.set(it.next().get() * it.next().get());

			String s = key.toString();
			newKey.set(s.substring(0, s.indexOf(",")));
//			System.out.println("NK: " + newKey.toString() + " -> " + result.get());
			context.write(newKey, result);
		}
	}

	public static class MVMReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
		private IntWritable result = new IntWritable();

		public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
//			System.out.println("kkkkkkk " + key);
			int sum = 0;
			for (IntWritable val : values) {
				sum += val.get();
//				System.out.println("vvvvvv " + val.get());
			}
			result.set(sum);
			context.write(key, result);
		}
	}

	public static class VectorPartitioner extends Partitioner<Text, IntWritable> {
		@Override
		public int getPartition(Text key, IntWritable value, int numPartitions) {
			double intervalSize = Math.ceil((double)length / numPartitions);
			
			return Integer.parseInt(key.toString()) / (int)intervalSize;
		}

	}

}

