package edu.calstatela.hadoop.example.associations;

import edu.calstatela.utils.MatrixCalculator;

import java.io.IOException;
import java.util.Collections;
import java.util.Iterator;
import java.util.StringTokenizer;
import java.util.Map;
import java.util.HashMap;
import java.util.List;
import java.util.LinkedList;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;

//import org.apache.hadoop.mapred.jobcontrol.Job;
import org.apache.hadoop.mapred.JobConf;

import org.apache.hadoop.conf.Configuration;

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.mapreduce.lib.output.SequenceFileOutputFormat;

import org.apache.log4j.Logger;

/**
 * Market Basket Analysis Algorithm: find the association rule for the list of items 
 * in a basket; That is, there are transaction data in a store
 * <ul>
 * <li>trx1: apple, cracker, soda, corn </li>
 * <l1>trax2: icecream, soda, bread</li>
 * <li>...</li>
 * <ul>
 * 
 * <p>
 * The code reads the data as 
 * key: first item
 * value: the rest of the items
 * 
 * <p>
 * And, count the possible associations as requested by user: two pairs, triples, ...
 * (see <a href="http://dal-cloudcomputing.blogspot.com">Jongwook's Map/Reduce blog</a>)
 * 
 * @date: 03/28/2011
 * @author jongwook Woo
 * @email: jwoo5@calstatela.edu
 * @version: 1.0
 *
 */
public class ItemCount extends Configured implements Tool {
	public static final Logger LOG = Logger.getLogger(ItemCount.class);
		
	public int run(String args[]) throws Exception {
		String inputPath = args[0];
		String outputPath = args[1];
		int noReducers = Integer.parseInt(args[2]);
		int noPairs = Integer.parseInt(args[3]);
		
		LOG.info("Input Path: " + inputPath);
		LOG.info("Output Path: " + outputPath);
		LOG.info("# of Reducers: " + noReducers);
		LOG.info("# of Pairs: " + noPairs);
		
		// job configuration
		// reference: 
		//http://www.hongliangjie.com/2011/01/16/passing-parameters-and-arguments-to-mapper-and-reducer-in-hadoop/
		// Create a new JobConf
	    JobConf jobConf = new JobConf(getConf(), ItemCount.class);	
	    jobConf.set("numPairs", args[3]);	
	    jobConf.setJobName("ItemCountJobConf");	
	    Job job = new Job(jobConf);

		job.setJarByClass(ItemCount.class);
		job.setNumReduceTasks(noReducers);
		//input/output path
		FileInputFormat.setInputPaths(job, new Path(inputPath));
		FileOutputFormat.setOutputPath(job, new Path(outputPath));		

		//Mapper K, V output
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);	
		//output format
		job.setOutputFormatClass(SequenceFileOutputFormat.class);
		
		//Reducer K, V output
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		// set mapper/reducer
		job.setMapperClass(MyMapper.class);
		// TODO: don't need combiner?
		job.setCombinerClass(MyReducer.class);
		job.setReducerClass(MyReducer.class);
		
		//delete the output path if exists to avoid "existing dir/file" error
		Path outputDir = new Path(outputPath);
		FileSystem.get(getConf()).delete(outputDir, true);
		
		long sTime = System.currentTimeMillis();
		job.waitForCompletion(true);
		System.out.println("It takes: " + (System.currentTimeMillis() - sTime) + " msec");
	
		return 0;
	}
		
	/**
	 * MyMapper
	 * input: (key1, value1)
	 * output: <(key2, value2)>
	 * @author jongwook
	 *
	 */
	protected static class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
		//output key2: list of items paired; can be 2 or 3 ...
		private static final Text itemList = new Text();
		//output value2: number of the paired items in the item list
		private static final IntWritable countItemList = new IntWritable(1);
		
		/**
		 * 
		 * build <key, value> by sorting the input list
		 * If not sort the input, it may have duplicated list but not considered as a same list.
		 * ex: (a, b, c) and (a, c, b) might become different items to be counted if not sorted
		 * @param no_pairs	number of pairs associated
		 * @param itr		input line
		 * @param context	Hadoop Job context
		 * @throws IOException
		 * @throws InterruptedException
		 */
		protected void buildOutputWithSort(int no_pairs, StringTokenizer itr, Context context) throws IOException, InterruptedException{
			//key is collected with an empty item list
			List<String> itemsST = new LinkedList<String>();			
			while (itr.hasMoreTokens()){ 
				itemsST.add((String)itr.nextToken());			
			}
			//sort items not to duplicate the items
			// ex: (a, b, c) and (a, c, b) might become different items to be counted if not sorted
			Collections.sort(itemsST);
			
			Iterator it = itemsST.iterator();
			List<String> items = new LinkedList<String>();	
			while (it.hasNext()){ 
				items.add((String)it.next());			
				// add the item list of the key
				List<List<String>> outItems = null;
				String itemPair = null;
				for(int i=0;i<items.size();i++){ 
					outItems = MatrixCalculator.selfCross(items, no_pairs);
					for(List<String> lst: outItems){
						itemPair = lst.get(0);// + " " + lst.get(1);
						for(int j=1;j<no_pairs;j++){
							itemPair += (" " + lst.get(j)); 
						}
						itemList.set(itemPair);
						context.write(itemList, countItemList);
					}
					
					
				}
			}
			
		}
		
		/**
		 * 
		 * build <key, value> from the input list
		 * It may have duplicated list but not considered as a same list.
		 * ex: (a, b, c) and (a, c, b) becomes different items to be counted
		 * 
		 * @param no_pairs	number of pairs associated
		 * @param itr		input line
		 * @param context	Hadoop Job context
		 * @throws IOException
		 * @throws InterruptedException
		 */
		protected void buildOutput(int no_pairs, StringTokenizer itr, Context context) throws IOException, InterruptedException{
			//key is collected with an empty item list
			List<String> items = new LinkedList<String>();
			
			while (itr.hasMoreTokens()){ 
				items.add((String)itr.nextToken());			
				// add the item list of the key
				List<List<String>> outItems = null;
				String itemPair = null;
				for(int i=0;i<items.size();i++){ 
					outItems = MatrixCalculator.selfCross(items, no_pairs);
					for(List<String> lst: outItems){
						itemPair = lst.get(0);// + " " + lst.get(1);
						for(int j=1;j<no_pairs;j++){
							itemPair += (" " + lst.get(j)); 
						}
						itemList.set(itemPair);
						context.write(itemList, countItemList);
					}
					
					
				}
			}
			
		}
		
		public void map(LongWritable key1, Text value1, Context context) throws IOException, InterruptedException{
			// take the value from the command line
			JobConf jobConf = (JobConf)context.getConfiguration();
			int no_pairs = Integer.parseInt(jobConf.get("numPairs"));
			LOG.info("# of Pairs for Mapper: " + no_pairs);

			// input line
			String line = value1.toString();
			StringTokenizer itr = new StringTokenizer(line);
			
			//buildOutput(no_pairs, itr, items, context);
			buildOutputWithSort(no_pairs, itr, context);
		}
		
		public void mapOld(LongWritable key1, Text value1, Context context) throws IOException, InterruptedException{
			// take the value from the command line
			JobConf jobConf = (JobConf)context.getConfiguration();
			int no_pairs = Integer.parseInt(jobConf.get("numPairs"));
			LOG.info("# of Pairs for Mapper: " + no_pairs);

			// input line
			String line = value1.toString();
			StringTokenizer itr = new StringTokenizer(line);
			
			//key is collected with an empty item list
			List<String> items = new LinkedList<String>();
			while (itr.hasMoreTokens()){ 
				items.add((String)itr.nextToken());			
				// add the item list of the key
				List<List<String>> outItems = null;
				String itemPair = null;
				for(int i=0;i<items.size();i++){ 
					outItems = MatrixCalculator.selfCross(items, no_pairs);
					for(List<String> lst: outItems){
						itemPair = lst.get(0);// + " " + lst.get(1);
						for(int j=1;j<no_pairs;j++){
							itemPair += (" " + lst.get(j)); 
						}
						itemList.set(itemPair);
						context.write(itemList, countItemList);
					}
					
					
				}
			}
		}		
		
	}
	
	
	/**
	 * MyReducer
	 * input: (key2, <value2>)
	 * output: <(key3, value3)>
	 * 
	 * @author jongwook
	 *
	 */
	protected static class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
		
		// output value3: total counted items paired
		private static final IntWritable totalItemsPaired = new IntWritable();
		
		public void reduce(Text key2, Iterable<IntWritable> values2, Context context) throws IOException, InterruptedException{
			int sum = 0;
			Iterator<IntWritable> iter = values2.iterator();
			while(iter.hasNext()){
				sum += iter.next().get();
			}
			totalItemsPaired.set(sum);
			context.write(key2, totalItemsPaired);
		}
	}
	
	
	private static int printUsage(){
		System.out.println("USAGE: [input-path] [output-path] [num-reducers] [num-pairs]");
		ToolRunner.printGenericCommandUsage(System.out);
		return -1;
	}

	// main to start from the command
	public static void main(String args[]) throws Exception {
		if(args.length != 4){
			printUsage();
			return;
		}

		System.exit(ToolRunner.run(new ItemCount(), args));
	}

}