一 什么是MapReduce

MapReduce是一种编程模型，也是一个处理和生成超大数据集的算法模型的相关实现。最早是由谷歌2004年公布的 MapReduce论文，论文描述了大数据的分布式计算方式，主要思想是将任务分解然后在多台处理能力较弱的计算节点中同时处理，然后将结果合并从而完成大数据处理。但是谷歌并没有把相关代码开源，是Doug Cutting根据谷歌的文章，实现了MapReduce，并将其作为Hadoop的一个组件开源[1]。

二 MapReduce解决了什么问题

在这个大数据时代，首先面对的问题就是数据的存储和计算。当数据量大到一定程度，一台机器已经无法存储了怎么办，那么最先想到的办法就是增加机器，让性能一般但是数量巨大的小型机器协同工作，最终的效果甚至能超越一个大型的工作站。那么怎么让这些小机器协同工作呢？MapReduce就是这样一个把这些机器组织起来的编程模型[2]。

在之前的主机时代，基本都是oracle，IBM这些大佬的天下。但是呀一个公司的命运，当然要靠自我奋斗，但也要考虑历史的进程。大量的创业小公司根本买不起oracle的产品，基本上是这样的情况：

硬件商：办过公司吗？

创业公司：没有

硬件商：我告诉你，办公司，得买机器，装机房，安数据库，拉拢关系，他们高兴了才能给你打折。以后用户给你交的钱，三七分成

创业公司：怎么才七成？

硬件商：七成是人家的，能得三成还得看 oracle的脸色

创业公司：谁的脸色？

硬件商：oracle

创业公司：oracle？我大老远的来一趟，就是为了看他的脸色？

硬件商：对

创业公司：我好不容易凑了钱，融了资，开了公司

硬件商：对

创业公司：我还要装机房

硬件商：对

创业公司：还要装数据库

硬件商：对

创业公司：还要拉拢关系

硬件商：对

创业公司：还要看他的脸色

硬件商：对

创业公司：那我不成了要饭的了吗

硬件商：那你要这么说，开公司还真就是跪着要饭的，就这，多少人想跪还没这门路呢！

创业公司：我问问你，我为什么要辞职创业

硬件商：不知道

创业公司：我就是腿脚不利索，跪不下去

硬件商：原来你是想站着挣钱，那你还是找工作去吧

创业公司：哎？那我就不明白了，我已经开了公司当老总，怎么还不如一个打工的？

硬件商：用户眼里，你是老总，可是在oracle眼里，你就是跪着要饭的。挣钱嘛，生意，不寒碜

创业公司：寒碜，很他妈寒碜

硬件商：不寒碜，你是想站着，还是想挣钱？

创业公司：我是想站着，还把钱挣了

硬件商：挣不成

创业公司：挣不成？

硬件商：挣不成

创业公司：这个，能不能把钱给挣了

硬件商：能挣，跪着

创业公司：这个能不能挣钱

硬件商：能挣，跪着

创业公司：这个加这个，能不能把钱给挣了

硬件商：敢问九筒大哥何方神圣？

创业公司：鄙人Google

三 MapReduce是怎么解决的

3.1 执行原理

通过将Map调用的输入数据自动分割为M个数据片段的集合，Map调用被分布到多台机器上执行。输入的数据片段能够在不同的机器上并行处理。使用分区函数将Map调用产生的中间key值分成R个不同分区（例如hash(key) mod R），Reduce调用也被分布到多台机器上执行。分区数量R和分区函数由用户来指定[3]。

图1 MapReduce过程

(1) 用户程序首先调用的 MapReduce 库将输入文件分成M个数据片（M是用户自定义），每个数据片段的大小一般从16MB到64MB(可以通过参数调节)。然后使用fork将用户程序拷贝到集群内其它机器上。

(2) 这些程序副本中的有一个特殊的程序——master。副本中其它的程序都是worker程序，master是负责调度的，为空闲worker分配作业（Map作业或者Reduce作业）。

(3) 被分配了Map任务的worker程序读取相关的输入数据片段，Map作业的数量和M值相等，和split对应，Map从输入的数据片段中解析出key/value pair，然后把key/value pair传递给用户自定义的Map函数，由Map函数生成并输出的中间key/value pair，并缓存在内存中。

(4) 缓存中的key/value pair通过分区函数分成R个区(由用户定义)，之后周期性的写入到本地磁盘上。缓存的key/value pair在本地磁盘上的存储位置将被回传给master，由master负责把这些存储位置再传送给Reduce worker。

(5) 当Reduce worker程序接收到master程序发来的数据存储位置信息后，使用RPC从Map worker所在主机的磁盘上读取这些缓存数据（可能不止一个地方，每个Map作业产生的中间键值对都可能映射到所有R个不同分区）。当Reduce worker读取了所有的中间数据后，通过对key进行排序后使得具有相同key值的数据聚合在一起。由于许多不同的key值会映射到相同的Reduce任务上，因此必须进行排序。如果中间数据太大无法在内存中完成排序，那么就要在外部进行排序。

(6) Reduce worker程序遍历排序后的中间数据，对于每一个唯一的中间key值，Reduce worker程序将这个key值和它相关的中间value值的集合传递给用户自定义的Reduce函数。Reduce函数的输出被追加到所属分区的输出文件。

(7) 当所有的Map和Reduce任务都完成之后，master唤醒用户程序。这时在用户程序里的对MapReduce调用才返回。

3.2 MapReduce过程细节

最近大火的电视剧《长安十二时辰》里频繁提到一个大扯淡术大案牍术，据我分析，这个就是MapReduce的原理。靖安司有全国大小的明档，密档，数量可谓汗牛充栋，要从这些档案中找到需要的信息，很明显不可能每个靖安吏都有徐宾那样的记忆力和计算力，所以就把一件大的工作分为小的子工作，下发给下面那些能力平常的小官员。

比如李必想要统计天宝二年每个人出入过大吉酒肆的次数，那么他作为提交一个程序给徐宾这个master：

Select user_name, count(1) num from DaJi_standing_book where created_date=’天宝二年’ group by user_name;

徐宾作为master，把统计任务下发给靖安吏，把靖安吏分为两部分，一部分我们姑且称之为Map吏，负责统计大吉酒肆的台账里的各个人名出现的数量；另一部分叫Reduce吏，负责汇总Map吏得到的结果。每个Map吏统计一部分原始的台账数据，每个Reduce吏负责统计部分人名的数据，Map吏统计完成后，Reduce吏从每个Map吏的结果中只拿自己所负责的人的数据，然后在汇总，最终得到结果返回给徐宾，再由徐宾把最终结果呈交给我们的李司丞。

它的详细过程是这样子的：

原始文档输入之后，首先对其进行切分，分成三片之后，对每个数据片启动三个Map作业，Map作业可以细分为4个阶段：record reader、mapper、combiner和partitioner[4]。

RecordReader

RecordReader表示以怎样的方式从分片中读取一条记录，每读取一条记录都会调用RecordReader类，系统默认的RecordReader是LineRecordReader，它是TextInputFormat对应的RecordReader；而SequenceFileInputFormat对应的RecordReader是SequenceFileRecordReader。LineRecordReader是每行的偏移量作为读入map的key，每行的内容作为读入map的value。很多时候hadoop内置的RecordReader并不能满足我们的需求，比如我们在读取记录的时候，希望Map读入的Key值不是偏移量而是行号或者是文件名，这时候就需要我们自定义RecordReader[5]。

Mapper

Mapper就是把输入的数据解析为键值对。

Combiner

Combiner和Partitioner是用来优化MapReduce的。可以提高MapReduce的运行效率。集群上的可用带宽限制了MapReduce作业的数量，因此尽量避免map和reduce任务之间的数据传输是有利的。Hadoop允许用户针对map任务的输出指定一个combiner（就像mapper,reducer）。combiner函数的输出作为reduce函数的输入。

Combiner可以看做局部的Reducer（local reducer）。它的作用是合并相同的key对应的value。在Mapper阶段，不管Combiner被调用多少次，都不应改变 Reduce的输出结果。Combiner通常与Reducer的逻辑是一样的，一般情况下不需要单独编写Combiner，直接使用Reducer的实现就可以了。

Combiner的优点是能够减少Map Task输出的数据量（即磁盘IO）。对spill，merge文件都可以进行压缩。Map Task输出越少，Reduce从Map结果中拉取的数据量就越少，自然就减少了网络传输的数据量。

但是并不是所有的场景都可以使用Combiner，必须满足结果可以累加。适合于Sum()求和，并不适合Average()求平均数。

Partitioner

Partitioner 处于 Mapper阶段，当Mapper处理好数据后，这些数据需要经过Partitioner进行分区，来选择不同的Reducer处理，从而将Mapper的输出结果均匀的分布在Reducer上面执行。

对于map输出的每一个键值对，系统都会给定一个partition，partition值默认通过计算key的hash值后对Reduce task的数量取模获得。如果一个键值对的partition值为1，意味着这个键值对会交给第一个Reducer处理。

Partitioner决定了Map Task输出的每条数据交给哪个Reduce Task来处理。Partitioner有两个功能：（1）均衡负载。它尽量将工作均匀地分配给不同的 Reduce。（2）效率。它的分配速度一定要非常快。Partitioner 的默认实现：hash(key) mod R，这里的R代表Reduce Task的数目，意思就是对key进行hash处理然后取模。

当Map过程结束后，进入reduce任务，其中reduce任务也可以分为4个阶段：混排（shuffle）、排序（sort）、reducer和输出格式（output format）

Shuffle

MapReduce中，mapper阶段处理的数据如何传递给reducer阶段，是MapReduce框架中最关键的一个流程，这个流程就叫Shuffle。

Shuffle的本义是洗牌、混洗，但MapReduce中的Shuffle更像是洗牌的逆过程，把一组无规则的数据尽量转换成一组具有一定规则的数据。

为什么MapReduce计算模型需要Shuffle过程？我们都知道MapReduce模型一般包括两个重要的阶段：Map是映射，负责数据的过滤分发；Reduce是规约，负责数据的计算归并。Reduce的数据来源于Map，Map的输出即是Reduce的输入，Reduce需要通过Shuffle来获取数据。

从Map输出到Reduce输入的整个过程可以广义地称为Shuffle。Shuffle横跨Map端和Reduce端，在Map端包括Spill过程，在Reduce端包括copy和sort过程，copy是从相应的Map节点中拉取需要的结果到Reduce中计算，一般Reduce是一边copy一边sort，即copy和sort两个阶段是重叠而不是完全分开的。

Reduce

最后Reduce规约，得到最终的结果。

看来Google不仅有优秀的程序员，还有优秀的历史学家，能从历史的故纸堆中恢复出这套大扯淡术大案牍术。

3.3 MapReduce容错

因为MapReduce的设计初衷是使用由成百上千的机器组成的集群来处理超大规模的数据，所以它必须要能很好的处理机器故障。

Worker故障

master周期性的ping每个worker。如果在一个约定的时间范围内没有收到worker 返回的信息，master将把这个worker标记为失效。所有由这个失效的worker完成的Map 任务被重设为初始的空闲状态，之后这些任务就可以被安排给其他worker。同样的，worker 失效时正在运行的Map或Reduce任务也将被重新置为空闲状态，等待重新调度。

Master故障

Master保有一些数据结构，它存储每一个 Map 和 Reduce 任务的状态（空闲、工作中或完成)，以及Worker机器(非空闲任务的机器)的标识。

一个简单的解决办法是让master周期性的将这些数据的写入磁盘，即检查点（checkpoint，对，就是游戏里的checkpoint）。如果这个master任务失效了，可以从最后一个检查点（checkpoint）开始启动另一个master进程。然而，由于只有一个master进程，master失效后再恢复是比较麻烦的，因此我们现在的实现是如果master失效，就中止MapReduce运算。客户可以检查到这个状态，并且可以根据需要重新执行MapReduce操作。

MapReduce是很不错，但是Master挂了，整个过程就GG了，而显然，电视剧中也是这样的，徐宾这个Master搞了点鬼，然后就自己而不是大案牍术选出了张小敬，后来把档案库烧了然后装死（涉及剧透），整个靖安司的大数据，啊不，大案牍集群就彻底完蛋干不了事情了。

四 MapReduce存在的问题

MapReduce是很不错，但是用来直接编程太不友好，它只有Map，Reduce两个方法，使用MapReduce对两个表join都是一项很有技巧的工作，更别说更复杂操作。比如我们的李司丞忽然灵xian机de一dan动teng，想要看同时去过怀远坊并在陇右当过兵的人的资料，我们来介绍其中一种最简单的join方案——reduce side join[7]，其主要过程是：
（1）在map阶段，把所有记录标记成<key, value>的形式，其中key是name，value则根据来源不同取不同的形式：来源于表A的记录，value的值为"a#"+所属团部；来源于表B的记录，value的值为"b#"+时间。

（2）在reduce阶段，先把每个key下的value列表拆分为分别来自表A和表B的两部分，分别放入两个向量中。然后遍历两个向量做笛卡尔积，形成一条条最终结果。

五 怎么解决这些问题

MR模型只有两个过程，编程较为困难，Tez是基于Hadoop Yarn之上的DAG（有向无环图，Directed Acyclic Graph）计算框架。它把Ｍap/Reduce过程再拆成若干个子过程，同时可以把多个Ｍap/Reduce任务组合成一个较大的DAG任务，减少了Ｍap/Reduce之间的文件存储。同时合理组合其子过程，也可以减少任务的运行时间[8]。

附录

[1] Hadoop之父Doug Cutting

[2] 关于MapReduce的理解？

[3] MapReduce: Simplifified Data Processing on Large Clusters

[4] 通俗易懂地理解MapReduce的工作原理

[5] MapReduce自定义RecordReader

[6] MapReduce优化——Combiner与Partitioner

[7] MapReduce：实现join的几种方法

[8] MapReduce和Tez对比

最后希望大家关注我的公众号：木南的算法笔记，爱你们哟~