MapReduce教程（01）- 初识MapReduce-CFANZ编程社区

文章目录

01 引言02 MapReduce概述

2.1 MapReduce定义
2.2 MapReduce工作流程
2.3 MapReduce流程对象

2.3.1 InputFormat
2.3.2 InputSplit
2.3.3 RecordReader
2.3.4 Mapper
2.3.5 Combiner
2.3.6 Partitioner
2.3.7 Shuffle 和排序
2.3.8 Reducer
2.3.9 OutputFormat

03 MapReduce原理图
04 文末

01 引言

在前面的《Hive教程》，我们知道Hive最终会使用MapReduce写入数据到HDFS的。

MapReduce教程（01）- 初识MapReduce_mapreduce

所以本专栏主要讲解MapReduce的相关内容。

02 MapReduce概述

2.1 MapReduce定义

MapReduce：是 Hadoop 生态下面的计算层，它把任务分割成小任务并分发到集群的机器上并行执行。

MapReduce教程（01）- 初识MapReduce_键值对_02

2.2 MapReduce工作流程

再把上图细分到代码的层面，MapReduce任务工作流程如下图：

MapReduce教程（01）- 初识MapReduce_数据_03

从上面的流程图，从左往右可以看到MapReduce的流程设计到几个对象，分别是：

输入文件：就是要存入HDFS 上的文件，格式有多种多样，比如有文本文件，二进制文件等；
InputFormat：是 MapReduce 框架的一个类，它对输入文件进行分割和读取，并创建数据分片 InputSplit；
InputSplit：数据分片对象，由InputFormat生成的，一个数据分片由一个 Mapper 来处理；
RecordReader：它会跟 InputSplit 交互，并把数据转换成适合mapper读取的键值对（key-value pair）记录，它会给文件的每一行数据分配一个字节偏移量（byte offset）作为唯一编号；
Mapper：负责处理每一个来自RecordReader的记录，并生成新的键值对数据，这些Mapper新生成的键值对跟输入键值对是不一样的，Mapper 的输出将会被写到本地磁盘;
Combiner：其实是一种 reduce 操作，它会对 mapper 的输出数据做本地聚合，也就是说它是在输出数据的 mapper所在的机器上执行的，主要为了减少 mapper和 reducer 之间的数据传输；
Partitioner：Partitioner 对来自combiner的输出数据分区并排序，其实就是对数据的 key 做哈希运算，具有相同key的记录会被分到相同的分区，然后每个分区会被发送给 reducer；
Shuffle 和排序：对数据进行跨网络的物理移动，需要消耗网络带宽资源。在所有 mapper 都完成之后，他们的输出数据才会被 shuffle 到 reduce 节点，并且这些 mapper产生的数据会被合并和排序，然后作为 reduce 阶段的输入数据；
Reducer：它把 mapper输出的键值对数据作为输入，然后对每个键值对数据记录应用 reducer函数并输出结果;
RecordWrite：负责把来自 Reducer 输出的键值对数据写到输出文件；
OutputFormat：将 Reducer输出的键值对写入到 HDFS 的，写入输出文件的方式由 OutputFormat决定。

下面对每个对象进行详细的描述。

2.3 MapReduce流程对象

2.3.1 InputFormat

InputFormat ：按照某个策略将输入数据且分成若干个 split数据分片，在 MapReduce框架中，一个split就意味着需要一个Map Task。

InputFormat分类：

【FileInputFormat】：指定数据文件所在的输入目录，启动一个 Hadoop 作业的时候，FileInputFormat 提供了一个包含要读取的文件的路径，它会读取路径下的所有文件，并把这些文件分割成一个或多个 InputSplit ；
【TextInputFormat】：把输入文件的每一个数据行当成一个记录，这对于无格式数据或者基于行数据文件（如日志文件）非常有用；
【KeyValueTextInputFormat】：和 TextInputFormat 类似，它也会把输入文件的每一个数据行当成一个记录；
【SequenceFileInputFormat】：用来读取顺序文件（sequence file）的，sequence 文件是块压缩的，并且提供了几种数据类型（不仅仅是文本类型）直接的序列化和反序列化操作，MapReduce作业间传输十分高效；
【SequenceFileAsTextInputFormat】：是 SequenceFileInputFormat 的另一种形式；
【SequenceFileAsBinaryInputFormat】：跟 SequenceFileAsTextInputFormat 类似，只是它从 sequence 文件读取 Key / Value 的时候是以二进制的形式读取的；
【NLineInputFormat】：是 TextInputFormat 的另一种形式，key 是数据行的字节偏移量，value 是数据行具体内容；
【DBInputFormat】：是一种使用 JDBC 从关系数据库读取数据的 InputFormat。使用 DBInputFormat 读取数据的时候， Key 是 LongWritables 类型，而 Value 是 DBWritables 类型

注意：可以定制化开发 InputFormat，来决定分割文件的方式。

2.3.2 InputSplit

InputSplit ：是数据的一种逻辑表示，即我们所说的文件的数据分片，每个分片由一个 mapper处理，并不真正存储输入数据，它只是数据的一种引用，即一种数据的逻辑表示。

我们并不需要直接处理 InputSplit，因为他们是由 InputFormat 创建的（ InputFormat 创建 InputSplit 并把它转换成键值对）。默认情况下，FileInputFormat把文件分割成若干个块，每个块大小 128MB（和 HDFS 的块大小一样），这个默认值可以在配置文件 mapred-site.xml 修改，参数为 mapred.min.split.size。也可以在提交作业的时候修改该值。

2.3.3 RecordReader

RecordReader ：读取由 InputSplit 创建的数据并创建相应的键值对。

RecordReader实例是被InputFormat定义的，默认情况下，它使用 TextInputFormat把数据转换成键值对。TextInputFormat提供 2 种RecordReader，分别是 :

【LineRecordReader 】：它把输入文件的每一行内容作为value，字节偏移量作为 key；
【SequenceFileRecordReader】：用于把二进制数据转换成记录形式。

一个被处理的记录的大小是有限制的，可以通过下面参数设置这个最大值：

conf.setInt("mapred.linerecordreader.maxlength", Integer.MAX_VALUE);

2.3.4 Mapper

Mapper：主要负责处理每个输入记录，并生成一个新键值对，这个键值对跟输入记录是完成不一样的，mapper 任务的输出数据由这些键值对组成的集合。在 mapper 任务把数据写到本地磁盘之前，数据会被按 key 进行分区并排序，分区的目的是要把 key相同的值聚集在一起。

MapReduce 框架为每个 InputSplit（数据分片）生成一个 map 任务。

map 任务数量 = {( 数据总容量 ) / ( 分片大小 )}

如果数据是 1TB，数据分片大小是 100MB 的话，那么 map 任务数 = ( 1000 * 1000 ) / 100 = 10000。即 10000 个 map。

2.3.5 Combiner

Combiner: 也被称为 “微型 reducer ”，主要工作就是在 Mapper 的输出数据被传输到Reducer之前对这些数据进行处理，在减少网络阻塞方面扮演着一个关键的角色。

2.3.6 Partitioner

Partitioner ：是用来对 mapper 输出的数据进行分区的，partitioner通过哈希函数对Key 或者 Key 的子集求哈希值，哈希值相同的键值对将在同一个分区里面。

2.3.7 Shuffle 和排序

Shuffle：指的是把map的输出结果从 Mapper 传输到 Reducer 的过程

排序：由 mapper生成的key会被MapReduce框架自动排序

2.3.8 Reducer

Reducer：以mapper输出的中间结果（键值对）作为输入数据，并对这些数据逐一执行 reducer 函数，比如可以对这些数据做聚合、过滤、合并等操作。Reducer首先按key处理键值对的值，然后生成输出数据（零或多个键值对），key 相同的键值对数据将会进入同一个reducer，并且 reducer是并行执行的，因为 reducer之间不存在依赖关系。

我们可以通过 Job.setNumreduceTasks(int)方法设置reduce的数量。一般合适的 reduce任务数量可以通过下面公式计算：

(0.95 或者 1.75) * ( 节点数 * 每个节点最大的容器数量)

使用 0.95 的时候，当 map 任务完成后，reducer会立即执行并开始传输 map的输出数据。使用 1.75 的时候，第一批reducer任务将在运行速度更快的节点上执行完成，而第二批reducer任务的执行在负载平衡方面做得更好。

2.3.9 OutputFormat

OutputFormat ：负责检验job的输出规范，RecordWriter把输出数据写到输出文件的具体实现就是由 OutputFormat决定的。

OutputFormat 的分类：

TextOutputFormat：它是以每一行一个键值对的形式写入数据的
SequenceFileOutputFormat：将它的输出写为一个顺序文件
SequenceFileAsBinaryOutputFormat：与 SequenceFileAsBinaryInputFormat 相对应，它以原始的二进制格式把键值对写到一个顺序文件中
MapFileOutputFormat：以 map 文件作为输出
MultipleOutput：将数据写到多个文件，这些文件的名称源于输出的键和值或者任意字符串；
LazyOutputFormat：是一个封装输出格式，可以保证指定分区第一条记录输出时才真正创建文件；
DBOutputFormat：它适用于将作业输出数据转储到关系数据库和 HBase。