MapReduce超详解

03-19 1009阅读

简介

概述

MapReduce是Hadoop提供的一套用于进行分布式计算的模型，本身是Doug Cutting根据Google的仿照实现的。

MapReduce超详解
（图片来源网络，侵删）

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依赖：



junit
junit
4.13.2



org.apache.logging.log4j
log4j-slf4j-impl
2.20.0



org.apache.hadoop
hadoop-common
3.2.4



org.apache.hadoop
hadoop-client
3.2.4



org.apache.hadoop
hadoop-hdfs
3.2.4

定义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 - 输出的值的类型。本案例中，输出的是个数，所以类型是LongWritable
public class CharCountMapper extends Mapper {
    // 次数
    private final LongWritable once = new LongWritable(1);
    // 需要覆盖map方法，将处理逻辑放入map方法中
    // key：键。行的字节偏移量
    // value：值，读取的一行数据
    // context：环境参数，可以利用这个参数将数据传递给ReduceTask
    @Override
    protected void map(LongWritable key, Text value, Mapper.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 {
    // 覆盖reduce方法，将逻辑写到reduce方法中
    // key：键。本案例中，是字符
    // values：值。本案例中，是字符对应的次数
    // context：环境参数
    @Override
    protected void reduce(Text key, Iterable values, Reducer.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 {
    private final IntWritable once = new IntWritable(1);
    @Override
    protected void map(LongWritable key, Text value, Mapper.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 {
    @Override
    protected void reduce(Text key, Iterable values, Reducer.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;
// 如果不需要值，那么值的类型可以是NullWritable
public class IPMapper extends Mapper {
    @Override
    protected void map(LongWritable key, Text value, Mapper.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 {
    @Override
    protected void reduce(Text key, Iterable values, Reducer.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);
    }
}

组件

Writable-序列化

在Hadoop的集群工作过程中，一般是利用RPC来进行集群节点之间的通信和消息的传输，所以要求MapReduce处理的对象必须可以进行序列化/反序列操作。Hadoop并没有使用Java原生的序列化，而是底层默认使用的序列化机制是AVRO。MapReduce针对常见的数据类型提供了其序列化形式：

表-1 序列化形式

Java类型

MapReduce类型

byte

Bytewritable

short

ShortWritable

int

IntWritable

long

LongWritable

floa
VPS购买请点击我

文章版权声明：除非注明，否则均为主机测评原创文章，转载或复制请以超链接形式并注明出处。

相关阅读

目录[+]