上一篇《大数据(3)：HDFS》分析了 Hadoop 的分布式存储框架 HDFS，这一篇将分析 Hadoop 的并行计算框架 MapReduce。
MapReduce 源于 Google 一篇论文。
2003 年，MapReduce 的诞生标志了超大规模数据处理的第一次革命，而开创这个时代的就是下面这篇论文：《MapReduce: Simplified Data Processing on Large Clusters》。
MapReduce 充分借鉴了“分而治之”的思想，将一个数据处理过程拆分为主要的 Map (映射) 与 Reduce (归约) 两步。简单地说，MapReduce 就是“任务的分解与结果的汇总”。
到了 2014 年左右，Google 内部已经几乎没人写新的 MapReduce 了。
你可能有一个疑问 ：为什么 MapReduce 会被取代？
事实是：MapReduce 是一种编程范式，更是一种思维方式。某个软件没有人用了，你可以说这个软件被淘汰了，但是你不能说实现这个软件的思想被淘汰了。
可以说原来用 Hadoop MapReduce 做的一些事情可以用其它引擎来做了，并不是 MapReduce 被淘汰了。所以，关于 MapReduce，我们更关注的应该是实现思想，而不是如何使用。
一、MapReduce 流程
先考虑一个最简单的 MapReduce 流程，主要分为 4 步：
-
input：从数据源提取数据，输入到 Map Task 中。
-
map：对输入片中的数据按照一定的规则映射成
(k, v)形式。 -
shuffle：这是一个比较核心的过程，shuffle 有洗牌的意思，eg. 把红桃都发给 A Reduce，黑桃都发给 B Reduce，诸如此类。
-
reduce：聚合，全部任务进行合并，即把分散的数据合并成一个大的数据，再对合并后的数据排序。
MapReduce process
例如使用 MapReduce 进行单词统计。
-
在 input 阶段：inputformat 按行读取文本，形成 (k, v) 形式，其中 k 是文本偏移量，v 是每行文本。
-
在 map 阶段：将上一步的 v，也就是每行文本，按空格分隔，分隔后形成新的 (k, v)，这时 k 是每个单词，v 是单词数量，目前每个单词数量都是 1，也就是 (word, 1) 这种形式。
-
在 shuffle 阶段，将上一阶段相同的 k 分组，例如所有的 (hello, 1) 将被分到同一组。
-
在 reduce 阶段，将每一组的结果聚合，例如某一组有 3 个 (tom, 1)，经过 reduce 聚合后变成 (tom, 3)。
形象的说：你需要统计一个文本里面有多少个单词。
- 把文本按行分割，每一行交个不同的人（input）。
- 每人计算自己手里有多少个单词（map）。
- 将相同的单词分到同一个组中（shuffle）。
- 你将所有组的结果综合起来，得到总的单词数量（reduce）。
但是这有几个问题，如果文本有一千行，而只有 3 个人，该怎么分配呢？在综合结果时，单词总数太多，你一个人忙不过来，怎么办？
二、InputFormat
在单词统计时，并不是帮你统计的人越多越好，也就是说在 MapReduce 程序的运行中，并不是 MapTask 越多就越好，需要考虑数据量的多少及机器的配置。
那么需要多少个 MapTask 呢？这取决于输入 InputFormat 的切片数量。InputFormat 默认的实现是按照 HDFS 的 block 大小 (128M) 进行切分 (split)。一个切片就对应一个 MapTask 实例。
假如我们有一个 300M 的文件，它会在 HDFS 中被切成 3 块：0 ~ 128M，128 ~ 256M，256 ~ 300M，然后将这 3 块放置到不同的节点上去（前提是节点数量大于 3，否则将以多线程方式运行）分别启动 MapTask。
也就是说 MapTask 的数量等于文件切片数量。
这是由于在实际过程中，如果 MapTask 读取的数据不在运行的本机，则须通过网络进行数据传输，这对性能的影响非常大。所以常采取的策略是就按照 block 块的存储切分 MapTask，使得每个 MapTask 尽可能读取本机的数据。
三、Reader
当文件进过 InputFormat 切分后，再通过 Reader 读入到 MapTask 中，这是由于 MapTask 的输入输出形式为 (k, v)，所以 Reader 的任务是将输入文件转成 (k, v)。
例如在单词统计时，输入是文本，Reader 将文本按行切割后，转成 (k, v)，其中 k 是文本偏移量，v 是每行文本。
四、MapTask
MapTask 主要分为以下几部分组成：map 函数、Partitioner、Combiner。
1. map 函数
在 map 函数中，接受 Reader 传来的每行文本，编程将每行文本按照空格分隔，形成新的 (k, v)，例如 (word, 1)。
2. Partitioner
分区。按照一定规则，把数据分成不同的区。
Partitioner 决定 MapTask 输出的数据交由哪个 ReduceTask 处理，也就是说 Partitioner 分区数 = Reduce 个数。
分区规则可通过编程自定义，默认是按照 key 的 hashcode 进行分区，相同 hash 值的 k 将被分到相同的区。同一个分区中的 (k, v) 最后将进入同一个 reduce。
但如果某个 k 大量存在，则它们的 hashcode 也相同，那么这些数据将会进入同一个 reduce 中，而其他的 reduce 中的数据很少，这将造成数据倾斜。
例如，有一万个单词需要统计，其中九千个是 hello，如果使用默认的分区方法，那么这九千个 hello 都将进入同一个 reduce 中，而其他 reduce 中的单词很少很少。
解决数据倾斜的一个常用办法是重写分区函数。
MapReduce
5. Combiner
Combiner 的作用顾名思义，就是聚合。不是 reduce 的作用就是聚合吗，怎么 Combiner 也聚合？
没错，Combiner 和 reduce 的作用很接近，都是聚合。举个例子，如果在某个 map 中，有一千个 (hello, 1)，那么这一千个都需要经过网络 shuffle 后，进入 reduce，这样大大增加了网络压力。
如果在网络 shuffle 之前，先在 map 中局部聚合，把一千个 (hello, 1) 变成一个 (hello, 1000)，这样一来，大大减小了网络压力。
但是 Combiner 不是万能的，例如在统计单词，或者统计最大最小值时，Combiner 是非常有效的，例如：
max(0, 20, 10, 25, 15) = max(max(0, 20, 10), max(25, 15)) = max(20, 25) = 25
但在统计平均值时，就不能使用 Combiner，例如：mean(0, 20, 10, 25, 15) = 14
但是：mean(mean(0, 20, 10), mean(25, 15)) = mean(10, 20) = 15
五、ReduceTask
ReduceTask 的任务是将 MapTask 的输出进行聚合。主要包括分组 (group) 和调用 reduce() 函数两部分。
分组，默认使用的是 WritableComparator 比较器进行对 key 值的比较，key 值相同的会被分在一组。而 reduce() 函数是按照组为操作对象进行统计的，也就是有多少个组，则调用几次 reduce() 函数。
一个完整的 MapReduce 流程如下：
MapReduce
网友评论