package org.apache.hadoop.examples;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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.GenericOptionsParser;
/**
 * 
 * 描述：WordCount explains by York
  * @author Hadoop Dev Group
 */
publicclass WordCount {
    /**
     * 建立Mapper類TokenizerMapper繼承自泛型類Mapper
     * Mapper類:實(shí)現(xiàn)了Map功能基類
     * Mapper接口：
     * WritableComparable接口：實(shí)現(xiàn)WritableComparable的類可以相互比較。所有被用作key的類應(yīng)該實(shí)現(xiàn)此接口。
     * Reporter 則可用于報(bào)告整個(gè)應(yīng)用的運(yùn)行進(jìn)度，本例中未使用。 
     * 
     */
  publicstaticclass TokenizerMapper 
       extends Mapper<Object, Text, Text, IntWritable>{
        /**
         * IntWritable, Text 均是 Hadoop 中實(shí)現(xiàn)的用于封裝 Java 數(shù)據(jù)類型的類，這些類實(shí)現(xiàn)了WritableComparable接口，
         * 都能夠被串行化從而便于在分布式環(huán)境中進(jìn)行數(shù)據(jù)交換，你可以將它們分別視為int,String 的替代品。
     * 聲明one常量和word用于存放單詞的變量
         */
    privatefinalstatic IntWritable one =new IntWritable(1);
    private Text word =new Text();
    /**
         * Mapper中的map方法：
         * void map(K1 key, V1 value, Context context)
         * 映射一個(gè)單個(gè)的輸入k/v對(duì)到一個(gè)中間的k/v對(duì)
         * 輸出對(duì)不需要和輸入對(duì)是相同的類型，輸入對(duì)可以映射到0個(gè)或多個(gè)輸出對(duì)。
         * Context：收集Mapper輸出的<k,v>對(duì)。
         * Context的write(k, v)方法:增加一個(gè)(k,v)對(duì)到context
         * 程序員主要編寫Map和Reduce函數(shù).這個(gè)Map函數(shù)使用StringTokenizer函數(shù)對(duì)字符串進(jìn)行分隔,通過write方法把單詞存入word中
     * write方法存入(單詞,1)這樣的二元組到context中
     */  
    publicvoid map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {
      StringTokenizer itr =new StringTokenizer(value.toString());
      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);
      }
    }
  }
  
  publicstaticclass IntSumReducer 
       extends Reducer<Text,IntWritable,Text,IntWritable> {
    private IntWritable result =new IntWritable();
    /**
         * Reducer類中的reduce方法：
      * void reduce(Text key, Iterable<IntWritable> values, Context context)
         * 中k/v來自于map函數(shù)中的context,可能經(jīng)過了進(jìn)一步處理(combiner),同樣通過context輸出           
         */
    publicvoid reduce(Text key, Iterable<IntWritable> values, 
                       Context context
                       ) throws IOException, InterruptedException {
      int sum =0;
      for (IntWritable val : values) {
        sum += val.get();
      }
      result.set(sum);
      context.write(key, result);
    }
  }

  publicstaticvoid main(String[] args) throws Exception {
        /**
         * Configuration：map/reduce的j配置類，向hadoop框架描述map-reduce執(zhí)行的工作
         */
    Configuration conf =new Configuration();
    String[] otherArgs =new GenericOptionsParser(conf, args).getRemainingArgs();
    if (otherArgs.length !=2) {
      System.err.println("Usage: wordcount <in> <out>");
      System.exit(2);
    }
    Job job =new Job(conf, "word count");    //設(shè)置一個(gè)用戶定義的job名稱
    job.setJarByClass(WordCount.class);
    job.setMapperClass(TokenizerMapper.class);    //為job設(shè)置Mapper類
    job.setCombinerClass(IntSumReducer.class);    //為job設(shè)置Combiner類
    job.setReducerClass(IntSumReducer.class);    //為job設(shè)置Reducer類
    job.setOutputKeyClass(Text.class);        //為job的輸出數(shù)據(jù)設(shè)置Key類
    job.setOutputValueClass(IntWritable.class);    //為job輸出設(shè)置value類
    FileInputFormat.addInputPath(job, new Path(otherArgs[0]));    //為job設(shè)置輸入路徑
    FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));//為job設(shè)置輸出路徑
    System.exit(job.waitForCompletion(true) ?0 : 1);        //運(yùn)行job
  }
}

public void map(Object key, Text value, Context context) throws IOException, InterruptedException {…}

public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {…}

Configuration conf = new Configuration();

    String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
    if (otherArgs.length != 2) {
      System.err.println("Usage: wordcount <in> <out>");
      System.exit(2);
    }

    Job job = new Job(conf, "word count");
    job.setJarByClass(WordCount.class);
    job.setMapperClass(TokenizerMapper.class);
    job.setCombinerClass(IntSumReducer.class);
    job.setReducerClass(IntSumReducer.class);

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

    FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
    FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);