package jp.ac.dendai.cdl.mori.wikie.mapred;

import java.io.IOException;

import jp.ac.dendai.cdl.mori.wikie.MapReduceDriver;
import jp.ac.dendai.cdl.mori.wikie.WikIE;
import jp.ac.dendai.cdl.mori.wikie.io.WikipediaXmlInputFormat;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.ParseException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class LangDriver implements MapReduceDriver {

    @Override
    public void configJob(Job job, String[] args) {
        job.setMapperClass(LangMapper.class);
        job.setReducerClass(LangReducer.class);

        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(Text.class);

        job.setInputFormatClass(WikipediaXmlInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        try {
            CommandLine commandLine = createCommandLine(args);
            job.getConfiguration().set("interwiki", commandLine.getOptionValue("interwiki"));
        } catch (ParseException e) {
            e.printStackTrace();
        }

    }

    @Override
    public int executeJob(Job job) throws IOException, InterruptedException,
            ClassNotFoundException {
        return job.waitForCompletion(true) ? 0: 1;
    }

    @Override
    public CommandLine createCommandLine(String[] args) throws ParseException {
        return WikIE.parseArgs(args);
    }
}
