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

2022-09-21 21:02:10

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