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如下：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>WordCount</groupId>
    <artifactId>Hadoop</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>2.7.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-core</artifactId>
            <version>2.7.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-common</artifactId>
            <version>2.7.0</version>
        </dependency>

    </dependencies>

</project>

编码

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<LongWritable, Text, Text, IntWritable> {

    // 直接把单词的个数设置成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<Text, IntWritable, Text, IntWritable> {

    private IntWritable result = new IntWritable();

    public void 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);
    }
}

编写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, R

log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout

log4j.appender.stdout.layout.ConversionPattern=%5p - %m%n

log4j.appender.R=org.apache.log4j.RollingFileAppender
log4j.appender.R.File=firestorm.log

log4j.appender.R.MaxFileSize=100KB
log4j.appender.R.MaxBackupIndex=1

log4j.appender.R.layout=org.apache.log4j.PatternLayout
log4j.appender.R.layout.ConversionPattern=%p %t %c - %m%n

log4j.logger.com.codefutures=DEBUG

配置Configuration如图：

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

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

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

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

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