import java.io.IOException;
import java.util.Locale;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
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.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * Zliczanie słów w „Panu Tadeuszu” - klasyczny MapReduce w Javie.
 *
 * Tokenizacja jest taka sama we wszystkich wersjach demo (Java, Streaming, Hive, Spark):
 * małe litery, podział na ciągi znaków niebędących literami Unicode, puste elementy odrzucone.
 *
 * Uruchomienie:  hadoop jar wordcount.jar WordCount WEJŚCIE_HDFS WYJŚCIE_HDFS
 */
public class WordCount extends Configured implements Tool {

    /** Mapper: dla każdej linii tekstu emituje pary (słowo, 1). */
    public static class WordMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        private static final Pattern NOT_LETTERS = Pattern.compile("[^\\p{L}]+");
        private static final IntWritable ONE = new IntWritable(1);
        private final Text word = new Text();

        @Override
        protected void map(LongWritable offset, Text line, Context context)
                throws IOException, InterruptedException {
            // małe litery, potem podział na wszystko, co nie jest literą
            String lower = line.toString().toLowerCase(Locale.ROOT);
            for (String token : NOT_LETTERS.split(lower)) {
                if (!token.isEmpty()) {
                    word.set(token);
                    context.write(word, ONE);
                }
            }
        }
    }

    /**
     * Combiner: „mały reducer” działający po stronie mappera. Sumuje częściowe wyniki
     * jeszcze przed wysłaniem ich siecią, więc przez shuffle płynie dużo mniej danych.
     */
    public static class SumCombiner extends Reducer<Text, IntWritable, Text, IntWritable> {
        private final IntWritable partial = new IntWritable();

        @Override
        protected void reduce(Text word, Iterable<IntWritable> counts, Context context)
                throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable count : counts) {
                sum += count.get();
            }
            partial.set(sum);
            context.write(word, partial);
        }
    }

    /** Reducer: dla każdego słowa sumuje wszystkie liczby od mapperów i combinerów. */
    public static class SumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private final IntWritable total = new IntWritable();

        @Override
        protected void reduce(Text word, Iterable<IntWritable> counts, Context context)
                throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable count : counts) {
                sum += count.get();
            }
            total.set(sum);
            context.write(word, total);
        }
    }

    @Override
    public int run(String[] args) throws Exception {
        if (args.length != 2) {
            System.err.println("Użycie: WordCount <katalog wejściowy HDFS> <katalog wyjściowy HDFS>");
            return 2;
        }
        Configuration conf = getConf();
        Job job = Job.getInstance(conf, "word count Pan Tadeusz (Java)");
        job.setJarByClass(WordCount.class);
        job.setMapperClass(WordMapper.class);
        job.setCombinerClass(SumCombiner.class);
        job.setReducerClass(SumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        // true = wypisuj postęp na konsoli (linie „map 0% reduce 0%”)
        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        System.exit(ToolRunner.run(new Configuration(), new WordCount(), args));
    }
}