package edu.jhu.thrax.hadoop.jobs;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
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 edu.jhu.thrax.hadoop.datatypes.AlignedRuleWritable;
import edu.jhu.thrax.hadoop.datatypes.Annotation;
import edu.jhu.thrax.hadoop.datatypes.RuleWritable;
import edu.jhu.thrax.hadoop.extraction.ExtractionCombiner;
import edu.jhu.thrax.hadoop.extraction.ExtractionMapper;
import edu.jhu.thrax.hadoop.extraction.ExtractionReducer;

public class ExtractionJob implements ThraxJob {

  public Set<Class<? extends ThraxJob>> getPrerequisites() {
    Set<Class<? extends ThraxJob>> result = new HashSet<Class<? extends ThraxJob>>();
    result.add(VocabularyJob.class);
    return result;
  }

  public Job getJob(Configuration conf) throws IOException {
    Job job = new Job(conf, "extraction");
    job.setJarByClass(ExtractionMapper.class);

    job.setMapperClass(ExtractionMapper.class);
    job.setCombinerClass(ExtractionCombiner.class);
    job.setReducerClass(ExtractionReducer.class);

    job.setSortComparatorClass(AlignedRuleWritable.RuleYieldComparator.class);
    job.setPartitionerClass(AlignedRuleWritable.RuleYieldPartitioner.class);
    
    job.setMapOutputKeyClass(AlignedRuleWritable.class);
    job.setMapOutputValueClass(Annotation.class);
    job.setOutputKeyClass(RuleWritable.class);
    job.setOutputValueClass(Annotation.class);

    job.setOutputFormatClass(SequenceFileOutputFormat.class);

    int numReducers = conf.getInt("thrax.reducers", conf.getInt("mapreduce.job.reduces", DefaultValues.DEFAULT_NUM_REDUCERS));
    job.setNumReduceTasks(numReducers);

    FileInputFormat.setInputPaths(job, new Path(conf.get("thrax.input-file")));
    int maxSplitSize = conf.getInt("thrax.max-split-size", 0);
    if (maxSplitSize != 0) FileInputFormat.setMaxInputSplitSize(job, maxSplitSize);

    FileOutputFormat.setOutputPath(job, new Path(conf.get("thrax.work-dir") + "rules"));

    return job;
  }

  // TODO: unify names of jobs and their output directories

  public String getName() {
    return "extraction";
  }

  public String getOutputSuffix() {
    return "rules";
  }
}
