package examples;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.UTF8;
import org.apache.hadoop.mapred.*;
import org.apache.lucene.analysis.SimpleAnalyzer;
import org.apache.lucene.analysis.Token;
import org.apache.lucene.analysis.TokenStream;

import java.io.IOException;
import java.io.StringReader;
import java.util.Iterator;

/**
 * Hadoop-compatible (old API) implementation of the RelWC algorithm as common example algorithm of the MRStreamer package.
 *
 * User: flangner
 * Date: 12.08.2010
 * Time: 10:28:50
 */
public class Wordcount implements Reducer<UTF8, IntWritable, UTF8, IntWritable>, Mapper<Object, Text, UTF8, IntWritable> {

    public void map(Object key, Text value, OutputCollector<UTF8, IntWritable> resultCollector, Reporter reporter) throws IOException {
        String line = value.toString();

        // b counts the bytes processed so far
        long b = line.getBytes().length;

        SimpleAnalyzer analyzer = new SimpleAnalyzer();

        TokenStream tokenStream = analyzer.tokenStream("", new StringReader(line));
        Token tok = tokenStream.next();
        while (tok != null) {
          resultCollector.collect (new UTF8(tok.termText()), new IntWritable(1));
          tok = tokenStream.next();
        }
        tokenStream.end();
        tokenStream.close();
    }

    public void reduce(UTF8 key, Iterator<IntWritable> values, OutputCollector<UTF8, IntWritable> collector, Reporter reporter) throws IOException {

        int value = 0;
        while (values.hasNext()) {
            value += values.next().get();
        }

        collector.collect(key, new IntWritable(value));
    }

    public void close() throws IOException { }

    public void configure(JobConf jobConf) { }

    public static void main(String[] args) throws IOException {
    JobConf conf = new JobConf(Wordcount.class);

    conf.setJobName("wordcount");
    conf.setOutputKeyClass(Text.class);
    conf.setOutputValueClass(IntWritable.class);
    conf.setMapperClass(Wordcount.class);
    conf.setCombinerClass(Wordcount.class);
    conf.setReducerClass(Wordcount.class);
    conf.setInputFormat(TextInputFormat.class);
    conf.setOutputFormat(TextOutputFormat.class);
    FileInputFormat.setInputPaths(conf, new Path("examples/testdata"));
    FileOutputFormat.setOutputPath(conf, new Path("examples/out"));

    conf.setNumMapTasks(4);
    conf.setNumReduceTasks(2);

    Long time = System.currentTimeMillis();
    MRSJobClient.runJob(conf);
    System.out.println("Duration of the execution of Relative Wordcount: " + (System.currentTimeMillis() - time) + " ms");
    }
}

