一个简单的mapreduce程序---统计每个单词出现次数_mapreduce统计单词出现次数

因为会对key进行排序，所以要把需要排序的元素写到key中

例如：统计单词出现次数，并按照次数降序。

           可以这样：map阶段输出<单词，1>，reduce阶段输出<单词，n>，map阶段输出<n,单词>，reduce阶段输出<单词，n>

统计每个单词出现次数

1.  定义一个Mapper类

2.  定义一个Reducer类

3.  定义一个Driver类(本机模拟运行)

附代码：

https://github.com/YanYuPS/wordCountSimple.git

//Map：每个单词记一个1
//输入key（一行文本的偏移量），输入value（一行文本内容），输出key（单词），输出value（单词个数）
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
 
   //MapTask调用map方法，每读取一个（keyIn，valueIn），就调用一次map方法
   @Override
   protected void map(LongWritable key, Text value, Context context)
         throws IOException, InterruptedException {
         
      //获取每一行的文本内容
      String lines = value.toString();
      String[] words = lines.split(" ");
      
      for (String word :words) {
         context.write(new Text(word), new IntWritable(1));
      } 
   }
}

//输入key（map的输出key：单词），输入value（map的输出value：单词个数），输出key（单词），输出value（单词个数）
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
 
   //ReduceTask调用reduce方法：会先将接收到的kv对按照key分组（key相同为一组）；然后每组key调用一次reduce方法
   @Override
   protected void reduce(Text key, Iterable<IntWritable> values,
         Context context) throws IOException, InterruptedException {
      
      int count =0;
      for(IntWritable v :values){
         count += v.get();
      }
      context.write(key, new IntWritable(count));
   }
}

//主类，指定job参数并提交job
public class WordCountDriver {
   
   public static void main(String[] args) throws Exception {
 
      Configuration conf = new Configuration();
      Job job = Job.getInstance(conf);
 
      //jar包位置
      job.setJarByClass(WordCountDriver.class);
      
      //指定mapper类和reducer类
      job.setMapperClass(WordCountMapper.class);
      job.setReducerClass(WordCountReducer.class);
      
      //map阶段，输出的数据类型
      job.setMapOutputKeyClass(Text.class);
      job.setMapOutputValueClass(IntWritable.class);
      //输出的数据类型
      job.setOutputKeyClass(Text.class);
      job.setOutputValueClass(IntWritable.class);
      
      //数据读取组件 输出组件
      job.setInputFormatClass(TextInputFormat.class);
      job.setOutputFormatClass(TextOutputFormat.class);
      
      //要处理的所有文件,所在文件夹路径
      FileInputFormat.setInputPaths(job, new Path("/Users/rainbow/Desktop/wordin"));
      
      //处理结果存放文件夹路径（如已存在，需删除）
      FileOutputFormat.setOutputPath(job, new Path("/Users/rainbow/Desktop/wordout"));
 
      //向yarn集群提交job
      boolean res = job.waitForCompletion(true);
      System.exit(res?0:1);
   }
}