MapReduce工作机制——Word Count实例（一）

MapReduce的思想是分布式计算，也就是分而治之，并行计算提高速度。

编程思想

首先，要将数据抽象为键值对的形式，map函数输入键值对，处理后，产生新的键值对作为中间结果输出。接着，MapReduce框架自动将中间结果按键做聚合处理，发给reduce函数处理。最后，reduce函数以键和对应的值的集合作为输入，处理后，产生另一系列键值对作为最终输出。后面会结合实例介绍整个过程。

运行环境

先不考虑采用YARN的情况，那个时候MapReduce的运行环境就是YARN，此处我们讨论的是上一代环境。

TaskTracker

slave的角色，负责汇报心跳和执行命令。一个集群有多个TaskTracker，但一个节点只有一个，TaskTracker和DataNode运行在同一节点。

JobTracker

master的角色，负责任务调度和集群资源监控，不参与计算。根据TaskTracker周期性发来的心跳信息，考虑TaskTracker的资源剩余量、作业优先级等等，为其分配合适的任务。

Word Count实例

环境

• Java 1.7
• Hadoop 2.7
• Maven 3.3
• Intellij IDEA 2016.3
• Windows 10

题主在集成开发环境下写了Word Count程序，配置的pom.xml如下：

    
        
     
      4.0.0
         
     
      WordCount
         
     
      Hadoop
         
     
      1.0-SNAPSHOT
         
             
                  
       
        org.apache.hadoop
                   
       
        hadoop-common
                   
       
        2.7.0
               
              
                  
       
        org.apache.hadoop
                   
       
        hadoop-mapreduce-client-core
                   
       
        2.7.0
               
              
                  
       
        org.apache.hadoop
                   
       
        hadoop-mapreduce-client-common
                   
       
        2.7.0
               
          
     
    

编码

Mapper类

Mapper类的4个泛型分别代表：map函数输入键值对的键的类，map函数输入键值对的值的类，map函数输出键值对的键的类，map函数输出键值对的值的类。

map函数部分，key是LongWritable类型，表示该行；value是Text类型，表示行的内容；Context类的write(Text key, IntWritable value)将中间结果输出。

package com.hellohadoop;import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;import java.util.StringTokenizer;/** * Created by duyue on 2017/7/13. */public class TokenizerMapper  extends Mapper
    
      {    // 直接把单词的个数设置成1, 认为出现了1次    private final static IntWritable one = new IntWritable(1);    private Text word = new Text();    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {        // 每行文本拆分成单个单词        StringTokenizer itr = new StringTokenizer(value.toString());        while (itr.hasMoreTokens()) {            word.set(itr.nextToken());            // 每个单词（忽略重复）的个数都为1            // 即，出现两次"good"会写入两次"good"，而不会认为"good"出现了2次            context.write(word, one);        }    }}
    

Reducer类

Reducer类的4个泛型表示：reduce函数输入键值对的键的类，reduce函数输入键值对的值的类（与map函数输出对应），reduce函数输出键值对的键的类，reduce函数输出键值对的值的类。

reduce函数部分：接收到的参数形如：<key, List<value>>，因为map函数把key值相同（同一单词）的所有value都发送给reduce函数，统计后输出结果。

package com.hellohadoop;import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;import java.util.Iterator;/** * Created by duyue on 2017/7/13. */public class IntSumReducer extends Reducer
    
      {    private IntWritable result = new IntWritable();    public void reduce(Text key, Iterable
     
       values, Context context) throws IOException, InterruptedException {        int sum = 0;        for (IntWritable val : values) {            sum += val.get();        }        result.set(sum);        context.write(key, result);    }}
     
    

编写main函数

package com.hellohadoop;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 java.io.IOException;/** * Created by duyue on 2017/7/13. */public class WordCount {    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {        Configuration conf = new Configuration();        Job job = new Job(conf, "word count");        job.setJarByClass(WordCount.class);        job.setMapperClass(TokenizerMapper.class);        job.setReducerClass(IntSumReducer.class);        job.setOutputKeyClass(Text.class);        job.setOutputValueClass(IntWritable.class);        FileInputFormat.addInputPath(job, new Path(args[0]));        FileOutputFormat.setOutputPath(job, new Path(args[1]));        System.exit(job.waitForCompletion(true) ? 0 : 1);    }}

运行程序

此处主要依赖于之前Maven依赖的包，为了成功显示日志文件，需要在resources包中添加log4j.properties文件，位置如下图：

文件配置：

log4j.rootLogger=debug, stdout, Rlog4j.appender.stdout=org.apache.log4j.ConsoleAppenderlog4j.appender.stdout.layout=org.apache.log4j.PatternLayoutlog4j.appender.stdout.layout.ConversionPattern=%5p - %m%nlog4j.appender.R=org.apache.log4j.RollingFileAppenderlog4j.appender.R.File=firestorm.loglog4j.appender.R.MaxFileSize=100KBlog4j.appender.R.MaxBackupIndex=1log4j.appender.R.layout=org.apache.log4j.PatternLayoutlog4j.appender.R.layout.ConversionPattern=%p %t %c - %m%nlog4j.logger.com.codefutures=DEBUG

配置Configuration如图：

自己要创建input文件夹，并将在Project Structure中设置为Excluded类型。在input文件夹下创建需要统计单词数的文件，位置如下图：

题主统计的是莎士比亚有名的十四行诗的Sonnet 18，运行程序后：

其中part-r-00000中保存了统计结果，图太长，截了一部分：

以上就是Word Count实例在Idea下运行的情况

下期预告：MapReduce编程涉及到的API