MapReduce超详解

简介

概述

MapReduce是Hadoop提供的一套用于进行分布式计算的模型，本身是Doug Cutting根据Google的<MapReduce: Simplified Data Processing on Large Clusters>仿照实现的。

MapReduce由两个阶段组成：Map(映射)阶段和Reduce(规约)阶段，用户只需要实现map以及reduce两个函数，即可实现分布式计算，这样做的目的是简化分布式程序的开发和调试周期。

特点

MapReduce的优点：

1)MapReduce易于编程：用户只需要简单的实现MapReduce提供的一些接口，就可以完成一个分布式程序，这个分布式程序可以分布到大量廉价的PC机器上运行。

2)具有良好的扩展性：当当前的集群的计算资源不能得到满足的时候，可以通过简单的增加机器来扩展它的计算能力。

3)高容错性：MapReduce设计的初衷就是使程序能够部署在廉价的PC机器上，这就要求它具有很高的容错性。例如，如果集群中某一台服务器宕机，那么MapReduce可以把上面的计算任务转移到另外一个节点上运行，不至于这个任务运行失败，而且这个过程不需要人工参与，而完全是由Hadoop内部完成的。

4)适合PB级以上海量数据的离线处理：可以实现上千台服务器集群并发工作，提供数据处理能力。

MapReduce的缺点：

1)不擅长实时计算：MapReduce的运行速度相对比较低，一般在毫秒或者秒级内返回结果，因此不适合于实时分析的场景。

2)不擅长流式计算：流式计算的输入数据是动态的，而MapReduce要求输入的数据集是静态的，不能动态变化。这是因为MapReduce自身的设计特点决定了数据源必须是静态的。

3)不擅长DAG(有向图)计算：多个应用程序存在依赖关系，后一个应用程序的输入为前一个的输出。在这种情况下，MapReduce并不是不能做，而是使用后，每个MapReduce作业的输出结果都会写入到磁盘，会造成大量的磁盘IO，导致性能非常的低下。

入门案例

思路

案例：统计一个文件中每一个字符出现的次数(处理文件：characters.txt)。

在MapReduce刚开始的时候，会先对文件进行切片(Split)处理。需要注意的是，切片本身是一种逻辑切分而不是物理切分，本质上就是在划分任务量，之后每一个切片会交给一个单独的MapTask来进行处理。默认情况下，Split和Block的大小是一致的。

切片之后，每一个切片(Split)会分配给一个单独的MapTask来处理。而MapTask确定好要处理的切片之后，默认情况下会对切片进行按行处理。需要注意，不同的MapTask之间只是处理的数据不同，但是处理的逻辑是相同的。

MapTask处理完数据之后，会将数据交给ReduceTask进行汇总。ReduceTask收到数据之后，会先将相同的键对应的值放到一组去，形成一个迭代器，这个过程称之为分组(group)。分组之后，再调用reduce方法对数据进行汇总处理，最终将处理结果写出到指定的文件系统中。

实现过程

导入POM依赖：

<dependencies><!--单元测试--><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.13.2</version></dependency><!--日志打印--><dependency><groupId>org.apache.logging.log4j</groupId><artifactId>log4j-slf4j-impl</artifactId><version>2.20.0</version></dependency><!--Hadoop通用包--><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>3.2.4</version></dependency><!--Hadoop客户端--><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>3.2.4</version></dependency><!--Hadoop HDFS--><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-hdfs</artifactId><version>3.2.4</version></dependency></dependencies>

定义Mapper类：

import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;// 需要继承Mapper类// 需要注意的是，MapReduce要求被处理的传输的数据能够被序列化// MapReduce提供了一套单独的序列化机制// KEYIN - 输入的键的类型。默认情况下，是行的字节偏移量// VALUEIN - 输入的值的类型。默认情况下，是输入的一行数据// KEYOUT - 输出的键的类型。本案例中，输出的是字符，所以类型是Text// VALUEOUT - 输出的值的类型。本案例中，输出的是个数，所以类型是LongWritablepublic class CharCountMapper extends Mapper<LongWritable, Text, Text, LongWritable> {    // 次数    private final LongWritable once = new LongWritable(1);    // 需要覆盖map方法，将处理逻辑放入map方法中    // key：键。行的字节偏移量    // value：值，读取的一行数据    // context：环境参数，可以利用这个参数将数据传递给ReduceTask    @Override    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {        // 获取一行数据        String line = value.toString();        // 拆分字符        char[] cs = line.toCharArray();        // 遍历数据，写出        for (char c : cs) {            context.write(new Text(String.valueOf(c)), once);        }    }}

定义Reducer类：

import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;// 需要继承Reducer// KEYIN,VALUEIN - 输入的键值类型。Reducer的数据从Mapper来，所以Mapper输出什么类型，Reducer就接收什么类型// KEYOUT,VALUEOUT - 输出的值的类型。本案例中，输出的是字符和次数public class CharCountReducer extends Reducer<Text, LongWritable, Text, LongWritable> {    // 覆盖reduce方法，将逻辑写到reduce方法中    // key：键。本案例中，是字符    // values：值。本案例中，是字符对应的次数    // context：环境参数    @Override    protected void reduce(Text key, Iterable<LongWritable> values, Reducer<Text, LongWritable, Text, LongWritable>.Context context) throws IOException, InterruptedException {        // 定义变量记录次数        long sum = 0;        // 遍历次数        for (LongWritable value : values) {            // 次数累计            sum += value.get();        }        // 写出结果        context.write(key, new LongWritable(sum));    }}

定义入口类(驱动类)：

import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.LongWritable;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;public class CharCountDriver {    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {        // 获取环境变量        Configuration conf = new Configuration();        // 获取任务        Job job = Job.getInstance(conf);        // 指定入口类        job.setJarByClass(CharCountDriver.class);        // 设置Mapper类        job.setMapperClass(CharCountMapper.class);        // 设置Reducer类        job.setReducerClass(CharCountReducer.class);        // 设置Mapper的输出的键的类型        job.setMapOutputKeyClass(Text.class);        // 设置Mapper的输出的值的类型        job.setMapOutputValueClass(LongWritable.class);        // 设置Reducer的输出的键的类型        job.setOutputKeyClass(Text.class);        // 设置Reducer的输出的值的类型        job.setOutputValueClass(LongWritable.class);        // 设置输入路径        FileInputFormat.addInputPath(job, new Path("hdfs://hadoop01:9000/txt/characters.txt"));        // 设置输出路径        FileOutputFormat.setOutputPath(job, new Path("hdfs://hadoop01:9000/result/char_count"));        // 提交任务，等待结束        job.waitForCompletion(true);    }}

在本地运行MapReduce之前，需要将Hadoop安装目录解压到本地的路径下，然后需要将给定资料中的bin.7z解压到相应的bin目录下，然后双击winutils.exe，如果是出现一个黑窗口一闪而过，则表示没有任何问题。如果双击winutils.exe出错，则需要将msvcr120.dll文件拷贝到C:\Windows\System32目录下，然后再双击winutils.exe。

之后需要配置环境变量：HADOOP_HOME，Path和HADOOP_USER_NAME。

如果在指定输入的时候，指定路径是一个目录，那么MapReduce会处理这个目录下的所有的文件。

问题解决

如果运行过程中出现了null/bin/winutils.exe，那么解决方案如下：

1)先检查环境变量是否配置正确；

2)如果环境变量正确，那么可以在Drivers中添加如下代码：

System.setProperty("hadoop.home.dir","Hadoop的解压路径");

如果运行过程中出现了NativeIO$Windows，那么解决方案如下：

1)先检查环境变量是否配置正确；

2)如果环境变量配置正确，那么可以将bin目录下的hadoop.dll文件拷贝到C:\Windows\System32目录下，再运行代码看是否配置正确；

3)如果上述方案依然无效，那么需要将给定资料中的NativeIO.java文件拷贝到当前工程下，建好对应的包。

练习

练习一：统计一个文件中单词出现的次数(处理文件：words.txt)。

Mapper类：

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;public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {    private final IntWritable once = new IntWritable(1);    @Override    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {        // 拆分单词        String[] arr = value.toString().split(" ");        // 遍历，写出        for (String s : arr) {            context.write(new Text(s), once);        }    }}

Reducer类：

import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {    @Override    protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {        // 定义变量记录和        int sum = 0;        // 遍历，求和        for (IntWritable value : values) {            sum += value.get();        }        // 写出        context.write(key, new IntWritable(sum));    }}

驱动类：

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;public class WordCountDriver {    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {        Configuration conf = new Configuration();        Job job = Job.getInstance(conf);        job.setJarByClass(WordCountDriver.class);        job.setMapperClass(WordCountMapper.class);        job.setReducerClass(WordCountReducer.class);        // 如果Mapper和Reducer的输出类型一致，可以只设置一次        job.setOutputKeyClass(Text.class);        job.setOutputValueClass(IntWritable.class);        FileInputFormat.addInputPath(job, new Path("hdfs://hadoop01:9000/txt/words.txt"));        FileOutputFormat.setOutputPath(job, new Path("hdfs://hadoop01:9000/result/word_count"));        job.waitForCompletion(true);    }}

练习二：IP去重(处理文件：ip.txt)。

Mapper类：

import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.NullWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;// 如果不需要值，那么值的类型可以是NullWritablepublic class IPMapper extends Mapper<LongWritable, Text, Text, NullWritable> {    @Override    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {        // 获取IP，写出        context.write(value, NullWritable.get());    }}

Reducer类：

import org.apache.hadoop.io.NullWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;public class IPReducer extends Reducer<Text, NullWritable, Text, NullWritable> {    @Override    protected void reduce(Text key, Iterable<NullWritable> values, Reducer<Text, NullWritable, Text, NullWritable>.Context context) throws IOException, InterruptedException {        context.write(key, NullWritable.get());    }}

驱动类：

import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.NullWritable;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;public class IPDriver {    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {        Configuration conf = new Configuration();        Job job = Job.getInstance(conf);        job.setJarByClass(IPDriver.class);        job.setMapperClass(IPMapper.class);        job.setReducerClass(IPReducer.class);                job.setOutputKeyClass(Text.class);        job.setOutputValueClass(NullWritable.class);        FileInputFormat.addInputPath(job, new Path("hdfs://hadoop01:9000/txt/ip.txt"));        FileOutputFormat.setOutputPath(job, new Path("hdfs://hadoop01:9000/result/ip"));        job.waitForCompletion(true);    }}