Hadoop系列-MapReduce工作原理详解（十二）

让我们详细的来讨论一下键值对 是怎么生成的。

InputSplit ：它是数据的逻辑表示，即数据分片，每个 InputSplit 会被一个 map 任务处理。
RecordReader ：它会跟 InputSplit 交互，并把数据转换成适合 mapper 读取的键值对（key-value pair）记录。默认情况下，它用的是 TextInputFormat 类来做转换。RecordReader 与 InputSplit 交互一直到文件读取完成。它会给文件的每一行数据分配一个字节偏移量（byte offset）作为唯一编号。后续这些键值对将被发送给 mapper 做进一步处理。

Hadoop Mapper 是如何工作的

下面让我们来看一下 hadoop 里面 mapper 的工作过程。

InputSplit 会为 mapper 将 HDFS 块转换成逻辑分片。比如，要读取的文件是 200MB，那么就需要 2 个 InputSplit，因为这个有文件有 2 个block（默认块大小为 128MB），MapReduce 会为每个块创建一个 InputSplit，然后为每个 InputSplit 创建一个 RecordReader 和一个 mapper。

InputSplit 的数量并非总是依赖于数据块的数量，我们可以通过设置下面的属性，来为文件指定分片的数量。

mapred.max.split.size

RecordReader 主要负责读取数据，并把数据转换成 的形式。它会为文件的每一行数据分配一个字节偏移量（唯一编号），这些键值对数据会被发送给 mapper，mapper 处理之后把数据落地到本地磁盘，我们把 mapper 输出的数据叫做中间数据，其实就是临时数据。

map 任务的数量

本小节我们来讨论一下在执行 MapReduce 作业的时候，对于给定的数据量，如何计算出 mapper 任务数量？一般情况下，输入文件的数据块总量决定了 mapper 任务的数量。对于 map 阶段，每个节点的最佳并行度在 10 到 100 个 mapper 之间，虽然对于非 CPU 密集型的 map 任务，并行度已经被设置为 300 个 map。由于任务的启动需要消耗一些时间，所以如果 map 的执行超过一分钟的话，任务运行效率会更高。

举个例子，假如 HDFS 的块大小是 128MB，需要处理的数据的大小是 10TB，那么就需要 82000 个 map。这个 map 的数量由 InputFormat 类决定，其计算公式如下：

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

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

二、MapReduce Reducer

Reducer 获取 Mapper 的输出数据（ 键值对 ），并对这些数据逐个处理，最终把最终处理结果存储到 HDFS。通常在 Hadoop 的 Reducer 阶段，我们会做聚合计算、求和计算等操作。

什么是 Hadoop Reducer

Reducer 处理 mapper 的输出数据，在处理这些数据的时候，它会生成一组新的输出数据。最后把数据存储到 HDFS。

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

MapReduce Reducer 阶段

通过前面章节的介绍我们已经知道，Hadoop MapReduce 的 Reducer 过程包含 3 个阶段。下面对他们进行逐个详细介绍。

Reducer 的 Shuffle 阶段

在本阶段，来自 mapper 的已经排好序的数据将作为 Reducer 的输入数据。在 Shuffle 阶段，MapReduce 通过 HTTP 协议拉取 mapper 输出数据的相应分区数据。

Reducer 的排序阶段

在该阶段，来自不同 mapper 的输入数据将会按 key 重新做排序。shuffle 和排序阶段是同时进行的。

Reduce 阶段

在 shuffle 和排序完成之后，reduce 任务对键值对数据做聚合操作。之后，OutputCollector.collect() 方法把 reduce 任务的输出结果写到 HDFS。Reducer 的输出不做排序。

Reducer 任务的数量

一个作业的 Reduce 任务数量是怎么确定的呢？以及如何修改 Reduce 数量？下面我们带着问题来详细了解一下。

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

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

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

三、MapReduce 键值对

Apache Hadoop 主要用于数据分析，我们利用数据分析里面的统计和逻辑技术来描述，说明和评估数据。Hadoop 可以用来处理机构化，非结构化和半结构化数据。在使用 Hadoop 的时候，当模式是静态的时候，我们可以直接使用模式的列，如果模式是非静态的，就不能用列的，因为没有列的概念了，只能使用键和值来处理数据。键和值并非是数据的固有属性，它只是在对数据做分析时人为选择的一种策略。

MapReduce 是 Hadoop 的核心组件，主要用于数据处理。Hadoop MapReduce 是一种软件框架，利用这个框架很容易就能开发出处理存储在 HDFS 的大数据量的应用程序。MapReduce 把处理过程分成两个阶段：Map 阶段和 Reduce 阶段。每个阶段的输入数据和输出数据都是以键值对的形式出现。

MapReduce 键值对的生成

让我们来了解一下 MapReduce 框架是怎么生成键值对的。MapReduce 过程中，在数据传输给 mapper 之前，数据首先被转换成键值对，因为 mapper 只能处理键值对形式的数据。

Hadoop MapReduce 生成键值对的过程如下：

InputSplit：这个是数据的逻辑表示形式。单个独立的 Mapper 处理的数据就是由 InputSplit 提供的。
RecordReader：它和 InputSplit 交互，并且把数据分片转换成适合 Mapper 读取的记录，这里的记录其实就是数据的键值对形式。默认情况下，RecordReader 使用 TextInputFormat 把数据转换成键值对。RecordReader 和 InputSplit 的交互直到文件读取完成才停止。

在 MapReduce 框架里面，map 函数处理某个键值对并发出一定数量的键值对，Reduce 函数处理按相同键分组的值，并发出另一组键值对作为输出。map 的输出类型应该和 reduce 的输入类型一致。比如：

• Map: (K1, V1) -> list (K2, V2)
• Reduce: {(K2, list (V2 }) -> list (K3, V3)

K2，V2 数据类型是要保持一致的。

Hadoop 中生成键值对需要依赖什么？

Hadoop 中键值对的生成取决于数据集和所需的输出。通常，键值对在4个位置指定：Map 输入、Map 输出、Reduce 输入和 Reduce 输出。

Map 输入

map 输入默认把数据文件的行数作为键（key），数据行对应的内容作为值。可以对 InputFormat 进行定制化开发，可以修改这个默认设置。

Map 输出

Map 主要负责过滤数据，并为数据基于键分组提供基础。

Key：它可能是字段、文本、对象，Reducer 端会根据 Key 对数据做分组和聚合。
Value：它可能是字段、文本、对象。它会被单独的 reduce 函数处理。

Reduce 输入

Map 的输出作为 Reduce的输入，也就是说，Reduce 的输入数据和 Map 输出数据是一样的。

Reduce 输出

这个取决于所需要的输出是怎么样的。

MapReduce 键值对举例

假如有个存储在 HDFS 的文件的内容如下：

John is Mark Joey is John

利用 InputFormat，我们就可以知道这个文件是怎么被分片和读取的。默认情况下，RecordReader 使用 TextInputFormat 把这个文件转换成键值对数据。

Key：这个是键值对的键，它是数据行的偏移量。
Value：值就是数据行具体内容，不包含行终止符。

根据上文件的内容，具体的 Key 和 Value 的内容如下：

Key 是 24
Value 是 John is Mark Joey is John

四、MapReduce InputFormat

Hadoop InputFormat 会检查作业的数据输入规范，它把输入文件分割成 InputSplit 分片，并发送给 Mapper。

Hadoop InputFormat

输入文件的分片和读取是由 InputFormat 定义的。InputFormat 主要负责创建数据分片，并把它转换成记录（即键值对），如果你还不熟悉 MapReduce 作业的工作原理，请参考  MapReduce 工作原理。

MapReduce 任务处理的数据是存储在输入文件的，而输入文件一般保存在 HDFS。虽然这些文件的格式可以是任意的，如基于行的日志文件和二进制格式都有可能被使用。利用 InputFormat 我们可以分割和读取这些文件。InputFormat 类是 MapReduce 框架最基础的类之一，它提供如下功能：

• 数据切分，按照某个策略将输入数据且分成若干个 split，以便确定 Map Task 的个数即 Mapper 的个数，在 MapReduce 框架中，一个 split 就意味着需要一个 Map Task。
• 为 Mapper 提供输入数据，即给定一个 split，使用其中的 RecordReader 对象将之解析为一个个的key/value 键值对。

如何把数据发送给 Mapper

MapReduce 框架有 2 种方法可以把数据发送给 mapper:getsplits() 和 createRecordReader() 。

public abstract class InputFormat<K, V>
{
public abstract List<InputSplit> getSplits(JobContext context)
throws IOException, InterruptedException;

public abstract RecordReader<K, V> createRecordReader(InputSplit split,
TaskAttemptContext context) throws IOException,
InterruptedException;

MapReduce InputFormat 的几种类型

让我们来看一下 Hadoop 里面的几种 InputFormat。

FileInputFormat

它是所有基于文件的 InputFormat 的基类。Hadoop FileInputFormat 指定数据文件所在的输入目录。当我们启动一个 Hadoop 作业的时候，FileInputFormat 提供了一个包含要读取的文件的路径，它会读取路径下的所有文件，并把这些文件分割成一个或多个 InputSplit 。

TextInputFormat

这个是 MapReduce 默认使用的 InputFormat。TextInputFormat 会把输入文件的每一个数据行当成一个记录，这对于无格式数据或者基于行数据文件（如日志文件）非常有用。

KeyValueTextInputFormat

KeyValueTextInputFormat 和 TextInputFormat 类似，它也会把输入文件的每一个数据行当成一个记录。TextInputFormat 是把数据行的内容作为键值对的 Key 部分，而 KeyValueTextInputFormat 分割 Key 和 value 是用 tab 键（‘\t’）来区分的。在一行数据里面，tab 键前面的内容作为 Key，tab 键后面的内容作为 Value。

SequenceFileInputFormat

Hadoop SequenceFileInputFormat 用来读取顺序文件（sequence file）的。sequence 文件是一种存储二进制键值对序列的二进制文件。sequence 文件是块压缩的，并且提供了几种数据类型（不仅仅是文本类型）直接的序列化和反序列化操作。sequence 文件可以作为 MapReduce 任务的输出数据，并且用它做一个 MapReduce 作业到另一个作业的中间数据是很高效的。

SequenceFileAsTextInputFormat

Hadoop SequenceFileAsTextInputFormat 是 SequenceFileInputFormat 的另一种形式，它把 sequence 文件的 key / value 转换成文本对象。通过调用 “toString()” 方法，就可以把 key / value 转换成文本对象（Text Object）。这种 InputFormat 使得 sequence 文件适合作为流输入。

SequenceFileAsBinaryInputFormat

跟 SequenceFileAsTextInputFormat 类似，只是它从 sequence 文件读取 Key / Value 的时候是以二进制的形式读取的。

NLineInputFormat

Hadoop NLineInputFormat 是 TextInputFormat 的另一种形式，key 是数据行的字节偏移量，value 是数据行具体内容。如果InputFormat 是 TextInputFormat 和 KeyValueTextInputFormat，那么每个 mapper 接收的输入行数是可变的，并且 mapper 的数量取决于分片的数量和数据行的长度。如果我们想要让 mapper 接收固定数量的输入行的话，我们可以使用 NLineInputFormat 。NLineInputFormat 前面的 N 表示每个Mapper收到输入的行数。默认 情况下（N=1），每个 mapper 刚好接收一行，如果 N = 2，那么每个分片是 2 行，一个 mapper 将接收两个键值对。

DBInputFormat

Hadoop DBInputFormat 是一种使用 JDBC 从关系数据库读取数据的 InputFormat。使用 DBInputFormat 读取数据的时候， Key 是 LongWritables 类型，而 Value 是 DBWritables 类型。

五、MapReduce InputSplit

Hadoop InputSplit

InputSplit 是数据的一种逻辑表示，即我们所说的文件的数据分片。每个分片由一个 mapper 处理，当然，mapper 并非直接对数据分片进行处理，在 mapper 处理数据分片之前，数据分片会被转换成记录，即键值对。mapper 直接操作键值对。

MapReduce InputSplit 的长度是以字节来度量的，每个 InputSplit 都有相应的存储位置（主机名称）。MapReduce 利用分片存储位置，把 map 任务调度到距离数据尽可能近的地方执行。Map 任务的执行优先级取决于分片的大小，分片大的最先执行，主要为了最小化作业的运行时间。需要注意的是，InputSplit，即数据分片并不真正存储输入数据，它只是数据的一种引用，即一种数据的逻辑表示。

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

Hadoop 如何修改数据分片大小

我们可以在 MapReduce 程序根据数据的大小来调整分片大小。修改 mapred.min.split.size 这个参数即可。

客户端（运行 Job 的节点）调用 getSplit() 方法为一个作业计算分片大小，然后发送到 application master，它根据数据分片的存储位置对处理该数据分片的 map 任务进行调度。接下来 map 任务把分片传递给 InputFormat 的 createRecordReader() 方法以便获取分片的 RecordReader 对象，而 RecordReader 对象将生成记录，也就是键值对，然后传递给 map 函数。

六、MapReduce RecordReader

为了理解 MapReduce 框架的 RecordReader 原理，首先得搞清楚 Hadoop 的数据流程。下面我来了解一下 Hadoop 里面的数据流程是怎样的。

Hadoop RecordReader 简介

MapReduce 有一个简单数据处理模型，map 和 reduce 函数的输入数据和输出数据都必须是键值对（key-value pairs）。Hadoop MapReduce 的 map 和 Reduce 函数具有以下通用的形式：

• map：(K1, V1) —> list(K2, V2)
• reduce：(K2, list(V2)) —> list(K3, V3)

在 MapReduce 运行之前，它必须要知道到底需要处理什么数据，这个过程由 InputFormat 类来完成，InputFormat 类主要负责从 HDFS 读取 map 函数需要处理的文件，并且它还会创建 InputSplit 并把 InputSplit 转换成记录。在 HDFS 数据会被分割成若干个块，块大小默认为 128 MB。默认每个分片对应一个块，由一个 map 任务处理。

InputFormat 类调用 getSplits() 函数计算每个文件的分片，然后把分片发送给 JobTracker，它利用分片的存储位置信息，把执行该分片的 map 任务调度到 TaskTracker。然后，Map 任务把分片传递给 TaskTracker 里面的 InputFormat类的 createRecordReader() 方法，以便获取分片的 RecordReader 对象。RecordReader 读取分片数据并把它转换成适合 mapper 读取的键值对形式。

Hadoop RecordReader 读取由 InputSplit 创建的数据并创建相应的键值对。“start” 是 RecordReader 开始生成键值对的起始字节位置，也就是一条记录的起始位置，“end” 是停止读取记录的标志，一条记录的结束位置。这里的 “start” 和 “end” 是由 InputSplit 确定的，“start” 通过 InputSplit.getStart() 获取，“end” 通过 InputSplit.getStart + InputSplit.length 获取。

RecordReader 工作原理

RecordReader 不仅仅是记录的迭代器。map 任务利用一个记录生成一个键值对，我们可以通过 mapper 的 run 函数验证这一点：

public void run(Context context) throws IOException, InterruptedException{
setup(context);
while(context.nextKeyValue())
{
map(context.getCurrentKey(),context.getCurrentValue(),context)
}
cleanup(context);
}

在执行 setup() 之后， nextKeyValue() 将重复从 context 获取下个记录来填充 mapper 的 key 和 value 对象。而 key 和 value 是通过 context 读取的。map 函数的输入数据是键值对形式的 (K, V)，当读取记录结束时，nextKeyValue 方法将返回 false。

RecordReader 根据 InputSplit 对数据划分好的数据边界进行读取，并生成键值对，但这个并非强制的，你可以通过自定义 RecordReader 来读取更多 InputSplit 之外的数据，但并不鼓励这么做。

RecordReader 分类

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

LineRecordReader

LineRecordReader 是 TextInputFormat 默认的 RecordReader。它把输入文件的每一行内容作为 value，字节偏移量作为 key。

SequenceFileRecordReader

这个是 SequenceFileInputFormat 对应的 RecordReader。用于把二进制数据转换成记录形式。

单个记录的最大值

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

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

如果一行数据的大小超过这个最大值，那么该记录会被忽略掉。

七、MapReduce Combiner

当我们用 MapReduce 作业处理大数据集的时候，Mapper 生成的中间结果数据就会比较大，而且这些中间结果数据后续会被传输到 Reducer 继续处理，这会导致非常大的网络开销，甚至网络堵塞。MapReduce 框架提供了一个函数——Hadoop Combiner 函数，它在减少网络阻塞方面扮演着一个关键的角色。

我们在之前已经学习了 Hadoop MapReduce 框架的 mapper 和 reducer。现在我们来学习 Hadoop MapReduce 框架的 Combiner。

MapReduce combiner 也被称为 “微型 reducer ”。combiner 的主要工作就是在 Mapper 的输出数据被传输到 Reducer 之前对这些数据进行处理。它在 mapper 之后 reducer 之前执行，也就是在 mapper 和 reducer 两个阶段的中间执行。并且 combiner 的执行是可选的，即可用可不用。

MapReduce combiner 工作原理

让我们来理解一下 Hadoop combiner 的工作原理，以及比较一下使用了 combiner 和未使用两者的区别。

未使用 combiner 的 MapReduce 程序

在上图中，MapReduce 程序没有使用 combiner。输入数据被分到两个 mapper 里面，并且生成了 9 个 key。现在 mapper 的中间结果已经产生了，即上图所示的 9 个键值对。后续 mapper 将会把这些数据直接发送到 reducer。在数据发送给 reducer 期间，这些数据的传输会消耗一些网络带宽（带宽，即在两台机器间传输数据的时间消耗）。如果数据容量很大，那么数据传输到 reducer 的耗时就更长。

如果我们在 mapper 和 reducer 中间使用了 combiner，那么，数据在从 mapper 传输到 reducer 之前，combiner 会对数据按 key 做聚合，那么输出的数据就是 4 个键值对。

使用 combiner 的 MapReduce 程序

在使用了 combiner 之后，Reducer 只需要处理由 2 个 combiner 生成的 4 个键值对数据。因此，reducer 生成最终结果只需要执行 4 次即可。作业的整体性能得到显著提升。

MapReduce combiner 优点

我们已经详细讨论了什么是 Hadoop MapReduce combiner，现在我们讨论一下使用 MapReduce combiner 的优势。

• Hadoop combiner 减少了数据从 mapper 到 reducer 之间的传输时间。
• 减少 reducer 处理的数据量，但最终计算结果不变。
• 提升 reducer 的整体性能。

MapReduce combiner 缺点

• MapReduce job 不能依赖于 Hadoop combiner 的执行，因为 combiner 的执行没有保证。
• 在本地文件系统中，键值对是存储在 Hadoop 中的，运行 combiner 会导致昂贵的磁盘IO 开销。

八、MapReduce Partitioner

MapReduce Partitioner 是用来对 mapper 输出的数据进行分区的。partitioner 通过哈希函数对 Key 或者 Key 的子集求哈希值，哈希值相同的键值对将在同一个分区里面。分区的数量取决于 reduce 任务的数量。下面对 MapReduce Partitioner 详细介绍。

MapReduce Partitioner

在开始学习 MapReduce Partitioner 之前，你需要先理解一下 mapper、reducer 以及 combiner 的概念。

map 输出的键值对的分区逻辑由 Partitioner 决定的。通过哈希函数，对 key 进行哈希并生成哈希值。每个 mapper 的输出都会被分区，key 相同记录将被分在同一个分区里。每个分区会被发送给一个 reducer。(key, value) 键值对将会被分到哪一个分区是由 Partition 类决定的。分区阶段发生在 map 阶段之后 reduce 阶段之前。MapReduce 为什么要设计分区这个阶段呢？下面会给出答案。

Hadoop MapReduce Partitioner 的必要性

现在让我们来讨论一下 Hadoop MapReduce Partitioner 的必要性。

MapReduce 作业接收一个输入数据集并生成键值对列表，这里的键值对列表是 map 阶段的输出结果数据，在这个阶段里面，输入数据会被切分成分片，每个 map 任务处理一个数据分片，并生成键值对列表。接着，map 的输出数据被发送到 reduce 任务，它会执行用户自定义的 reduce 函数，该函数会对 map 的输出数据进行计算。但在 reduce 阶段之前，map 的输出数据会被基于 Key 分区并排序。

分区的目的是把具有相同 key 的值集合在一起，确保 key 相同的值都会在同一个 reducer 里面。这样才能保证 map 的输出数据被均匀的分发到 reducer 。

默认的 MapReduce Partitioner

Hadoop MapReduce 默认的 Hadoop Partitioner 是哈希 Partitioner（Hash Partitioner），它会对 key 计算哈希值，并基于该哈希值对键值对数据进行分区。

一个 MapReduce job 会有多少个 Partitioner

Partitioner 的数量等于 reducer 的数量，Partitioner 会根据 reducer 的数量来划分数据，reducer 数量可以通过下面的方法进行设置：

JobConf.setNumReduceTasks()

因此，来自同一个分区的数据会被一个 reducer 任务处理。需要注意的是，只有作业具有多个reducer 任务时，分区才会被创建。也就是说，如果作业只有 1 个 reducer 任务，分区阶段是不会发生的。

低效的分区

如果在输入数据集里面，有一个 key 的数量比其他 key 的数量要大得多，这种情况，有两种机制可以对数据进行分区。

• 数量比其他 key 大得多的数据分到一个分区
• 其他 key 根据他们的 hashCode() 的值进行分区

但如果 hashCode() 方法不能均匀的把其他 key 的数据分发到分区，那么数据将会被均匀的发送到 reducer 任务。低效的分区意味着，某些 reducer 将比其他 reducer 任务处理更多的数据。那么，整个作业的运行时间将取决于这些需要处理更多数据的 reducer，也就是说，作业的运行时间会更长。

如何克服低效分区

为了克服低效分区的问题，我们可以自定义分区器（partitioner），这样我们就可以根据具体业务修改分区逻辑，把数据均分的分发到不同的 reducer 任务里。 

九、MapReduce Shuffle 和排序

mapper 的输出结果被传输到 reducer 的过程被称为 shuffle，并且在传输到 reducer 之前，数据会被按 key 排序。下面会详细介绍这两个过程。

Hadoop MapReduce Shuffle 和排序

在学习 shuffle 和排序之前，可以先复习一下 MapReduce 的其他阶段，比如  Mapper， Reducer， Combiner， partitioner 以及  InputFormat。

MapReduce 框架的 Shuffle 阶段指的是把 map 的输出结果从 Mapper 传输到 Reducer 的过程。MapReduce 的排序阶段包括对 map 输出的合并和排序两个步骤。mapper 的输出数据会被按 key 分组和排序。每一个 reducer 获取所有具有相同 key 的值。MapReduce 框架的 shuffle 和排序阶段是同时发生的。

Shuffle

我们都知道，数据从 mapper 传输到 reducer 的过程被称为 shuffle。所以，MapReduce Shuffle 阶段对于 reducer 来说是有必要的，否则，reducer 就没任务输入数据。由于 shuffle 可以在 map 阶段完成之前就启动，所以这回节省一些运行时间，以更少的时间完成任务。

排序

由 mapper 生成的 key 会被 MapReduce 框架自动排序，比如，在 reducer 启动之前，所有由 mapper 生成的中间结果键值对会按 key 进行排序而不是按 value 排序。传输给 reducer 的 value 并没有排序，它们的顺序是任意的。

Hadoop 中的排序帮助 reducer 轻松区分何时应该启动新的 reduce 任务。这会为 reducer 节省一些时间。在已排序的输入数据中，当下一个 key 和 前一个 key 不一样的时候，reducer 就会启动一个新的 reduce 任务。每个 reduce 任务以键值对作为输入并输出键值对数据。

需要注意的是，如果你通过下面的代码把 reducer 任务的个数设置为0，那么作业就不会有 shuffle 和排序阶段了。

setNumReduceTasks(0)

这时，MapReduce 作业在 map 阶段完成后就结束的，而且，map 阶段不包括任何类型的排序操作，所以这样的 map 阶段执行速度会更快。

MapReduce 二次排序

如果你想对 reducer 的键值对的 value 进行排序，那么，就需要用到二次排序技术了，二次排序技术可以让我们对传输到每个 reducer 的键值对的 value 以降序或者升序排序。

十、MapReduce OutputFormat

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

Hadoop Outputformat

在开始学习 MapReduce 框架的 OutputFormat 之前，让我们先来看一下 RecordWriter ，以及它在 MapReduce 框架起到什么样的作用。

Hadoop RecordWriter

我们知道，Reducer 以 mapper 输出的键值对结果为输入数据，并对这些数据执行 reducer 函数，它最终输出的结果依然是键值对数据。RecordWriter 负责把 Reducer 输出的键值对数据写到输出文件。

Hadoop Outputformat

在前面我们已经了解到，Hadoop RecordWriter 获取 Reducer 输出的数据，并把这些数据写到输出文件。而 RecordWriter 用怎样的格式把数据写到文件，这些行为是由 OutputFormat 决定的。OutputFormat 和 InputFormat 函数两者之间有许多共同之处。OutputFormat 函数负责把文件写到 HDFS 或者本地磁盘，它描述了 MapReduce job 输出数据的规范：

• MapReduce job 检查输出目录是否已经存在
• OutputFormat 提供的 RecordWriter 把结果数据写到输出文件，该输出文件会被存储在文件系统。

FileOutputformat.setOutputPath() 方法用来设置输出目录。每个 Reducer 会把数据写到公共目录下的一个独立文件中，每个 reducer 对应一个文件。

Hadoop OutputFormat 的分类

和 InputFormat 一样，Hadoop OutputFormat 也分好几种不同分类。

TextOutputFormat

Reducer 默认的输出格式是 TextOutputFormat，它是以每一行一个键值对的形式写入数据的，key 和 value 可能是任意数据类型的，但最终 TextOutputFormat 会调用 toString() 方法把他们转换成字符串类型。键值对之间以 tab 键作为分割符，你可以通过修改下面的属性修改这个默认的设置：

MapReduce.output.textoutputformat.separator

与 TextOutputFormat 对应的输入格式是 KeyValueTextInputFormat，它通过可配置的分隔符将键值对文本行分割。

SequenceFileOutputFormat

SequenceFileOutputformat 将它的输出写为一个顺序文件。如果输出需要作为后续 MapReduce 任务的输入，这是一种比较好的输出格式，因为它的格式紧凑，很容易被压缩。压缩由 SequenceFileOutputformat 的静态方法 putCompressionType() 来实现。

SequenceFileAsBinaryOutputFormat

SequenceFileAsBinaryOutputFormat 与 SequenceFileAsBinaryInputFormat 相对应，它以原始的二进制格式把键值对写到一个顺序文件中。

MapFileOutputFormat

MapFileOutputformat 以 map 文件作为输出。MapFile 中的键必须顺序添加，所以必须确保 reducer 输出的键是已经拍好序的。

MultipleOutput

MultipleOutput 类可以将数据写到多个文件，这些文件的名称源于输出的键和值或者任意字符串。这允许每个 reducer（或者只有 map 作业的 mapper ）创建多个文件。采用 name-m-nnnnn 形式的文件名用于 map 输出，name-r-nnnnn 形式的文件名用于 reduce 输出，其中 name 是由程序设定的任意名字，nnnnn 是一个指明块号的整数（从 00000 开始）。块号保证从不同分区（mapper 或者 reducer）写的输出在相同名字情况下不会冲突。

LazyOutputFormat

FileOutputformat 的子类会产生输出文件（ part-r-nnnnn ），即使文件是空的。有些应用倾向于不创建空文件，此时 LazyOutputFormat 就有用武之地了。它是一个封装输出格式，可以保证指定分区第一条记录输出时才真正创建文件。要使用它，用 JobConf 和相关的输出格式作为参数来调用 setOutputFormatClass() 方法即可。

DBOutputFormat

它适用于将作业输出数据转储到关系数据库和 HBase。

附1： MapReduce InputSplit 与 HDFS 块区别

InputSplit 即数据分片，HDFS 块（block）即分布式存储系统的数据块概念。下面详细介绍这两个概念的区别和联系。

HDFS 块与 InputSplit

HDFS 块

块是硬盘上存储数据的一个连续位置。通常，文件系统将数据存储成块的集合。同样的方式，HDFS 以块的方式存储文件。Hadoop 应用程序负责在多个节点分配数据块。

InputSplit

InputSplit 即我们所说的数据分片，一个单独的 mapper 处理的数据由 InputSplit 提供，即一个数据分片对应被一个 mapper 处理，数据分片会转换成记录，每个记录（即键值对）会被 map 处理。map 任务的个数等于数据分片的数量。

一开始 MapReduce 任务处理的数据是存储在输入文件的，而输入文件一般在 HDFS 。输入文件如何被读取和切分由 InputFormat 类来决定，另外它还负责创建 InputSplit。

InputSplit 和 块的比较

让我们来讨论 MapReduce InputSplit 和块之间的特性比较。

InputSplit 与块的大小比较

块：HDFS 块的默认大小是 128MB，我们可以按实际需求对该值进行配置。除了最后一个数据块，文件的其他所有块大小都是一样的，最后一个数据块大小一般小于等于 128MB。文件被切分成若干个大小为 128MB 的数据块，并存储到 Hadoop 文件系统。

InputSplit：默认情况下，InputSplit 的大小近似等于块大小。InputSplit 是由用户自己定义的，并且基于数据的容量可以在 MapReduce 程序调整 InputSplit 的值。

InputSplit 和 块的数据表示

块：块是数据的物理表示，它包含可读或者可写的最小数据量。
InputSplit：它是块中数据的逻辑表示。它是在 MapReduce 程序处理数据的时候被使用的。InputSplit 本身并不存储物理数据，它只是数据的一种引用。

InputSplit 和 块示例

假如我们需要把文件存储到 HDFS。HDFS 以块的形式存储文件，块是数据读取和存储的最小单位，并且块的默认大小是 128MB 。HDFS 把文件切分成块，并把块存储在集群的不同机器节点上，假如我们有一个 130MB 的文件，那么 HDFS 会把这个文件切割成 2 个块，如上第一个图所示。

现在，如果我们想对这些块执行 MapReduce 程序，那么它是不会被处理的，因为第二个块并不是完整的块。但是这个问题 InputSplit 可以解决。InputSplit 可以把一组 block 作为一个单独的块，因为 InputSplit 里面包含下一个块的位置以及完整的块所需的数据的字节偏移量。

附2：MapReduce 只有 Map 阶段的 job

在 Hadoop，只有 Map 任务的作业就是 mapper 处理了所有的数据处理任务，作业的整个过程没有 reducer 参与，而 mapper 的输出结果就是作业的最终输出结果。

仅 Map 阶段的 MapReduce 作业

MapReduce 是一种软件框架，利用该框架可以快速开发出处理大数据量的应用程序。MapReduce 框架的两个重要任务是：Map 任务和 Reduce 任务。Map 阶段接收一批数据，并把这些数据转换键值对。Reduce 阶段以 map 阶段的输出结果为输入数据，并且对这些数据基于 key 做聚合。

从上图我们可以知道，在一个 MapReduce job 里面有两组并行进程，即 map 和 reduce；在 map 进程，首先输入数据会被分发到多个 map 节点 ，然后每个单词被标记并生成键值对 。

第一个 mapper 节点接收到 3 个单词分别是 lion，tiger 和 river，因此节点的输出将会是 3 个键值对，其中 key 是单词本身，value 设置为 1。其他节点的处理过程跟第一个 mapper 一致。这些键值对之后会被传递给 reducer 节点，这样就会发生 shuffle 操作，也就是说，所有具有相同 key 的键值对会被分发到相同的节点。因此，在 reduce 进程中，就是对具有相同 key 的 value 进行运算操作。

现在让我们来考虑这样一个场景，如果我们只需要执行分发的操作，而不需要聚合的操作，那么这种情况下，我们就会更倾向于使用只有 map 阶段的作业。 在仅发生 Map 阶段的 Hadoop 作业中，map 阶段利用 InputSplit 就完成了所有的任务，并不需要 reducer 的参与。这里 map 的输出数据就是作业的最终输出结果了。

如何避免 Reduce 阶段的产生

为了避免触发 Reduce 任务，你可以在驱动程序（driver program）通过调用下面的方法来把 reduce 任务数设置为 0。

job.setNumreduceTasks(0)

设置完之后，Hadoop 作业在执行的时候就只有 map 阶段，而不会发生 reduce 了。

优点

在 map 和 reduce 两个阶段之间，包含排序和 shuffle 阶段。排序和 shuffle 负责对 key 升序排序和基于相同 key 对 value 进行分组。这个阶段的开销是非常大的。如果 reduce 阶段不是必要的，那么我们可以通过把 reduce 任务数设置为 0 来避免 reduce 阶段的发生，而且排序和 shuffle 阶段也不会发生。同时还可以避免数据传输的网络开销。

在一般的 MapReduce 作业中，mapper 的输出在被发送给 reducer 之前会先把输出数据写到本地磁盘，但在只有 map 的作业里，map 的输出会被直接写到 HDFS，这将节省了作业运行时间和 reduce 的开销。另外 partitioner 和 combiner 这两步在只有 map 的作业里是不需要的，所以这也会让作业运行的更快。

 参考： MapReduce Mapper | MapReduce 教程 ...

MapReduce InputSplit 与 HDFS 块 | MapReduce 教程