package HiBench;

import java.io.IOException;
import java.net.URISyntaxException;
import java.util.Iterator;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.SequenceFileOutputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapred.lib.NLineInputFormat;

public class PagerankData {

	private static final Log log = LogFactory.getLog(PagerankData.class.getName());
	
	private DataOptions options;

	private static final String VERTICALS_DIR_NAME = "vertices";
	private static final String EDGES_DIR_NAME = "edges";	
	private boolean balance = false;  //	original PAGERANK_NODE_BALANCE

	private String cdelim = "\t";

	private Dummy dummy;

	PagerankData(DataOptions options) {
		this.options = options;
		parseArgs(options.getRemainArgs());
	}

	private void parseArgs(String[] args) {
		
		for (int i=0; i<args.length; i++) {

			if ("-d".equals(args[i])) {
				cdelim = args[++i];
			} else if ("-pbalance".equals(args[i])) {
				balance = true;
			} else {
				DataOptions.printUsage("Unknown pagerank data arguments -- " + args[i] + "!!!");
			}
		}
	}
	
	public void init() throws IOException {
		
		log.info("Initializing PageRank data generator...");
		
		Utils.checkHdfsPath(options.getResultPath(), true);
		Utils.checkHdfsPath(options.getWorkPath(), true);

		Utils.serialLinkZipf(options);

		dummy = new Dummy(options.getWorkPath(), options.getNumMaps());
	}

	private void setPageRankNodesOptions(JobConf job) {
		job.setLong("pages", options.getNumPages());
		job.setLong("slotpages", options.getNumSlotPages());
	}
	
	private void setPageRankLinksOptions(JobConf job) throws URISyntaxException {
		job.setLong("pages", options.getNumPages());
		job.setLong("slotpages", options.getNumSlotPages());
		job.set("delimiter", cdelim);
		
		Utils.shareLinkZipfCore(options, job);
	}
	
	public static class BalancedLinkNodesMapper extends MapReduceBase implements
	Mapper<LongWritable, Text, LongWritable, NullWritable> {

		@Override
		public void map(LongWritable key, Text value,
				OutputCollector<LongWritable, NullWritable> output, Reporter reporter) throws IOException {
	
			String delimiter = "[ \t]";
			String[] pair = value.toString().split(delimiter);
			
			output.collect(
					new LongWritable(Long.parseLong(pair[0])),
					NullWritable.get()
					);
		}
	}

	public static class BalancedLinkNodesReducer extends MapReduceBase implements
	Reducer<LongWritable, NullWritable, NullWritable, Text> {

		@Override
		public void reduce(LongWritable key, Iterator<NullWritable> values,
				OutputCollector<NullWritable, Text> output, Reporter reporter)
						throws IOException {
	
			output.collect(NullWritable.get(), new Text(key.toString()));
		}
	}

	public static class DummyToNodesMapper extends MapReduceBase implements
	Mapper<LongWritable, Text, LongWritable, Text> {
		
		private long pages, slotpages;

		private void getOptions(JobConf job) {
			pages = job.getLong("pages", 0);
			slotpages = job.getLong("slotpages", 0);
		}

		@Override
		public void configure(JobConf job) {
			getOptions(job);
		}

		@Override
		public void map(LongWritable key, Text value,
				OutputCollector<LongWritable, Text> output, Reporter reporter) throws IOException {
	
			int slotId = Integer.parseInt(value.toString().trim());
			long[] range = HtmlCore.getPageRange(slotId, pages, slotpages);
			
			for (long i=range[0]; i<range[1]; i++) {
				key.set(i);
				Text v = new Text(Long.toString(i));
				output.collect(key, v);
				reporter.incrCounter(HiBench.Counters.BYTES_DATA_GENERATED, 8+v.getLength());
			}
		}
	}

	private void createPageRankNodesDirectly() throws IOException {

		log.info("Creating PageRank nodes...", null);

		Path fout = new Path(options.getResultPath(), VERTICALS_DIR_NAME);
		
		JobConf job = new JobConf(PagerankData.class);
		String jobname = "Create pagerank nodes";

		job.setJobName(jobname);
		setPageRankNodesOptions(job);

		job.setOutputKeyClass(LongWritable.class);
		job.setOutputValueClass(Text.class);
		
		FileInputFormat.setInputPaths(job, dummy.getPath());
		job.setInputFormat(NLineInputFormat.class);
		
		if (balance) {
			/***
			 * Balance the output order of nodes, to prevent the running
			 * of pagerank bench from potential data skew
			 */
			job.setMapOutputKeyClass(LongWritable.class);
			job.setMapOutputValueClass(NullWritable.class);
			
			job.setMapperClass(BalancedLinkNodesMapper.class);
			job.setReducerClass(BalancedLinkNodesReducer.class);
//			job.setPartitionerClass(ModulusPartitioner.class);

			if (options.getNumReds() > 0) {
				job.setNumReduceTasks(options.getNumReds());
			} else {
				job.setNumReduceTasks(Utils.getMaxNumReds());
			}
		} else {
			job.setMapOutputKeyClass(Text.class);
			job.setMapperClass(DummyToNodesMapper.class);
			job.setNumReduceTasks(0);
		}

		if (options.isSequenceOut()) {
			job.setOutputFormat(SequenceFileOutputFormat.class);
		} else {
			job.setOutputFormat(TextOutputFormat.class);
		}
		
		if (null != options.getCodecClass()) {
			job.set("mapred.output.compression.type","BLOCK");
			job.set("mapreduce.output.fileoutputformat.compress.type","BLOCK");
			FileOutputFormat.setCompressOutput(job, true);
			FileOutputFormat.setOutputCompressorClass(job, options.getCodecClass());
		}
		
		FileOutputFormat.setOutputPath(job, fout);

		log.info("Running Job: " +jobname);
		log.info("Dummy file " + dummy.getPath() + " as input");
		log.info("Vertices file " + fout + " as output");
		JobClient.runJob(job);
		log.info("Finished Running Job: " + jobname);
	}

	public static class DummyToPageRankLinksMapper extends MapReduceBase implements
	Mapper<LongWritable, Text, LongWritable, Text> {

		private static final Log log = LogFactory.getLog(DummyToPageRankLinksMapper.class.getName());
		private HtmlCore html;
		private long pages, slotpages;
		private String delim;

		private void getOptions(JobConf job) {
			pages = job.getLong("pages", 0);
			slotpages = job.getLong("slotpages", 0);
			delim = job.get("delimiter");
		}

		public void configure(JobConf job) {

			try {
				html = new HtmlCore(job);
				
				getOptions(job);
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	
		public void map(LongWritable key, Text value, OutputCollector<LongWritable, Text> output,
				Reporter reporter) throws IOException {

			int slotId = Integer.parseInt(value.toString().trim());
			html.fireRandom(slotId);

			long[] range = HtmlCore.getPageRange(slotId, pages, slotpages);

			/**
			 * For output collect
			 */
			for (long i=range[0]; i<range[1]; i++) {
				key.set(i);
				
				long[] linkids = html.genPureLinkIds();
				for (int j=0; j<linkids.length; j++) {
					String to = Long.toString(linkids[j]);
					Text v = new Text(to);
					output.collect(key, v);
					reporter.incrCounter(HiBench.Counters.BYTES_DATA_GENERATED, 8+v.getLength());
				}
				
				if (0==(i % 10000)) {
					log.info("still running: " + (i - range[0]) + " of " + slotpages);
				}
			}
		}
	}

	private void createPageRankLinksDirectly() throws IOException, URISyntaxException {

		log.info("Creating PageRank links", null);

		JobConf job = new JobConf(PagerankData.class);
		String jobname = "Create pagerank links";

		Path fout = new Path(options.getResultPath(), EDGES_DIR_NAME);

		job.setJobName(jobname);
		setPageRankLinksOptions(job);

		job.setOutputKeyClass(LongWritable.class);
		job.setOutputValueClass(Text.class);
//		job.setMapOutputKeyClass(LongWritable.class);
//		job.setMapOutputValueClass(Text.class);

		job.setNumReduceTasks(0);
		
		FileInputFormat.setInputPaths(job, dummy.getPath());
		job.setInputFormat(NLineInputFormat.class);

		job.setMapperClass(DummyToPageRankLinksMapper.class);

		if (options.isSequenceOut()) {
			job.setOutputFormat(SequenceFileOutputFormat.class);
		} else {
			job.setOutputFormat(TextOutputFormat.class);
		}
		
		if (null != options.getCodecClass()) {
			job.set("mapred.output.compression.type","BLOCK");
			job.set("mapreduce.output.fileoutputformat.compress.type", "BLOCK");
			FileOutputFormat.setCompressOutput(job, true);
			FileOutputFormat.setOutputCompressorClass(job, options.getCodecClass());
		}
		
		FileOutputFormat.setOutputPath(job, fout);
		
		log.info("Running Job: " +jobname);
		log.info("Dummy file " + dummy.getPath() + " as input");
		log.info("Edges file " + fout + " as output");
		JobClient.runJob(job);
		log.info("Finished Running Job: " + jobname);
	}

	public void generate() throws IOException, URISyntaxException {
		
		log.info("Generating pageRank data files...");
		init();
		createPageRankNodesDirectly();
		createPageRankLinksDirectly();
		closeGenerator();
	}

	private void closeGenerator() throws IOException {

		log.info("Closing pagerank data generator...");
		Utils.checkHdfsPath(options.getWorkPath(), true);
	}
}
