package com.bigdata.exercise1_2;


import java.io.IOException;
import java.util.Arrays;
import java.util.StringTokenizer;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
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.output.FileOutputFormat;
import org.apache.hadoop.conf.Configuration;


public class App {

	 public static class PreverjanjeSodostiLihosti extends Mapper<Object, Text, Text, IntWritable>
	 {
		 private final static IntWritable one = new IntWritable(1);
		 private final static IntWritable zero = new IntWritable(0);

		 private Text text = new Text();
		 private String word1 = new String();
		 private String word2 = new String();
		 private String word3 = new String();

		 public void map(Object key, Text value, Context context) throws IOException, InterruptedException
		 {
			 Configuration conf = context.getConfiguration();
			 conf.setAllowNullValueProperties(true);
			 word1 = conf.get("word1");
			 word2 = conf.get("word2");
			 word3 = conf.get("word3");

			 StringTokenizer itr = new StringTokenizer(value.toString());
			 while (itr.hasMoreTokens())
			 {
				 text.set(itr.nextToken());
				 if(text.toString().equals(word1)||text.toString().equals(word2)||text.toString().equals(word3))
				 {
					 context.write(text, one);
				 }
				 else if (!text.toString().equals(word1)||!text.toString().equals(word2)||!text.toString().equals(word3))
				 {
					 if(conf.onlyKeyExists(word1))
					 {
						 Text text1 = new Text(word1);
						 context.write(text1, zero);
					 }
					 if(conf.onlyKeyExists(word2))
					 {
						 Text text2 = new Text(word2);
						 context.write(text2, zero);
					 }
					 if(!word3.equals(null))
					 {
						 Text text3 = new Text(word3);
						 context.write(text3, zero);
					 }
				 }
			 }
		 }
	 }


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

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

public static void main(String... args) throws Exception {

Configuration conf = new Configuration();
conf.set("mapred.job.queue.name", "default");
conf.set("word1",args[2]);
for(int i = 3;i<args.length;i++)
{
	conf.set("word"+(i-1), args[i]);
}

// configuration should contain reference to your namenode
FileSystem fs = FileSystem.get(new Configuration());
// true stands for recursively deleting the folder you gave
fs.delete(new Path(args[1]), true);


Job job = Job.getInstance(conf, "Stetje Sodih in lihih stevil");

job.setJarByClass(App.class);
job.setMapperClass(PreverjanjeSodostiLihosti.class);
job.setReducerClass(SestejSodeInLihe.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);

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

System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}