// See
//   http://d.hatena.ne.jp/stanaka/20091125/1259124272
//   http://publish.viirya.org/?q=node/12

package hadoop

// import shadoop._
import java.util.Iterator
import java.io.IOException;
import java.util.StringTokenizer;

import java.util.Iterator
import org.apache.hadoop.fs._
import org.apache.hadoop.io._
import org.apache.hadoop.mapreduce._
import org.apache.hadoop.mapreduce.Mapper
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat
import org.apache.hadoop.conf.Configuration

object WordCount {

    class TokenizerMapper extends Mapper[LongWritable, Text, Text, LongWritable] {

        val one = new LongWritable(1)

        override def map(ky: LongWritable, value: Text, output: Mapper[LongWritable, Text, Text, LongWritable]#Context) = {
            val itr = new StringTokenizer(value.toString)
            while (itr.hasMoreTokens) {
                output.write(new Text(itr.nextToken), one)
            }
        }
    }

    class SumReducer extends Reducer[Text, LongWritable, Text, LongWritable] {
        override def reduce(key: Text, values: java.lang.Iterable[LongWritable],
                            output: Reducer[Text, LongWritable, Text, LongWritable]#Context) = {
            var sum = 0L
            val it = values.iterator
            while(it.hasNext) {
                sum += it.next.toString.toLong
            }
            output.write(key, new LongWritable(sum))
        }
    }

    def main(args: Array[String]): Unit = {
        val conf = new Configuration()
        val job = new Job(conf, "word count")

        job setJarByClass(WordCount getClass())

        job setMapperClass(classOf[WordCount.TokenizerMapper])
        job setCombinerClass(classOf[WordCount.SumReducer])
        job setReducerClass(classOf[WordCount.SumReducer])

        job setMapOutputKeyClass(classOf[Text])
        job setMapOutputValueClass(classOf[LongWritable])
        job setOutputKeyClass(classOf[Text])
        job setOutputValueClass(classOf[LongWritable])

        FileInputFormat addInputPath(job, new Path("input"))
        FileOutputFormat setOutputPath(job, new Path("output"))
        System exit(job waitForCompletion(true) match { case true => 0 case false => 1})

    }
}
