实验二-2 Eclipse&Hadoop 做英文词频统计进行集群测试

package twoWordCount;

import java.io.IOException;

import java.util.StringTokenizer;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

public class myMapper extends Mapper<LongWritable, Text, Text,   IntWritable> {

   private final static   IntWritable one = new IntWritable(1);     

    private Text word = new   Text();

   public void   map(LongWritable ikey, Text value, Context context)

         throws IOException,   InterruptedException {

      /*

       *将字符串解析成Key-Value的形式

       * ikey      偏移量

       * value     内容

       * context 上下文

       */

           StringTokenizer tokenizer = new   StringTokenizer(value.toString());

           while (tokenizer.hasMoreTokens()) {

               word.set(tokenizer.nextToken());

               context.write(word, one);

           }

     }

}

package twoWordCount;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

public class myReducer extends Reducer<Text, IntWritable,   Text, IntWritable> {

   private IntWritable   result = new IntWritable();

   public void reduce(Text   _key, Iterable<IntWritable> values, Context context)

         throws IOException,   InterruptedException {

      /*

       * 获取map方法的Key—Value结果，相同的Key发送到同一个Reducer里面，迭代Key，把value想加，结果写到HDFS系统中

       * _key       map端输出的key值

       * values     Map端输出的value集合(相同key值的集合)

       * context    Reducer端的上下文

       */

      int sum=0;

      for (IntWritable val :   values) {

            sum=sum+val.get();

         }

         result.set(sum);

         context.write(_key,result);

      }

}

package twoWordCount;

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.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import org.apache.hadoop.util.GenericOptionsParser;

public class myDriver {

   public static void   main(String[] args) throws Exception {

//    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);

//     }

       Configuration conf = new Configuration();

       String[] otherArgs = new   GenericOptionsParser(conf, args).getRemainingArgs();

       if (otherArgs.length < 2) {

         System.err.println("Usage:   wordcount <in> [<in>...] <out>");

         System.exit(2);

       }

      Job job = Job.getInstance(conf,   "EnglishWordCount");

      job.setJarByClass(twoWordCount.myDriver.class);

      // TODO: specify a   mapper

      job.setMapperClass(twoWordCount.myMapper.class);  //Mapper

      job.setCombinerClass(twoWordCount.myReducer.class);//作业合成类

      // TODO: specify a   reducer

      job.setReducerClass(twoWordCount.myReducer.class);//Reducer

      // TODO: specify   output types

      job.setOutputKeyClass(Text.class);                //设置作业输出数据的关键类

      job.setOutputValueClass(IntWritable.class);              //设置作业输出值类

      // TODO: specify input   and output DIRECTORIES (not files)

//    FileInputFormat.setInputPaths(job,   new Path(otherArgs[0]));  //文件输入

//    FileOutputFormat.setOutputPath(job,   new Path(otherArgs[1])); //文件输出

//

//    if   (!job.waitForCompletion(true))                  //等待完成输出

//       return;

      for (int i = 0; i <   otherArgs.length - 1; ++i) {

            FileInputFormat.addInputPath(job, new   Path(otherArgs[i]));

          }

          FileOutputFormat.setOutputPath(job,

            new Path(otherArgs[otherArgs.length -   1]));

          System.exit(job.waitForCompletion(true) ?   0 : 1);

   }

}