继续MapReduce

好，看完WordCount项目，继续来深入了解一下MapReduce。

1

我们先来看看MapReduce在Yarn上的工作流程。

首先有一个client（客户端），然后由客户端发送一个请求，这里的请求就是我要计算某个数据，把请求发送给ResourceManager。这里ResourceManager下面有两个东西，一个叫做Applications Manager（应用），一个叫Resource Scheduler（资源）。接着由Applications Manager寻找一个节点，并开启这个节点上面的Application Master，开启后，这个App Mstr要计算这个任务需要多少资源，并且把这些需求汇报给ResourceManager中的Applications Manager，接着ResourceManager中的Resource Scheduler就开始在机架中寻找有资源的机器。

找到以后就让App Mstr跑到对应的机器上面开启了Map Task（Map类和map方法）和Reduce Task（Reduce类和reduce方法）。

各个节点要实时地汇报任务执行的情况，最后全部都跑完之后，要最后汇报给Application Manager成功或者失败。

在8088端口上看到的效果就很直观了。

任务还没开始运行，资源的管理都是0。

任务一开始运行，就开始占用资源。

最最后成功，会直接反馈SUCCEED。

2

在项目中我也提到，并不是直接把Map处理完的数据马上传入给Reduce处理。中间还经历了一段过程，我后面补充说，在Reduce Task中还有一个group方法来处理数据，转化成Iterator。

那么实际上，在Map Task中也有一个类似的操作，我们把这两个操作整合到一起，起名叫Shuffle。

于是就有分成Map中的Shuffle和Reduce中的Shuffle。我们结合一下官网的图片来理解一下。

先说 Map Task

首先在对input进行切片，然后放到map方法运行，结束以后，下一个操作是“buffer in memory”，也就是在内存中开辟了一个空间，让数据不断写入。下一步，“partition，sort and spill to disk”，分区，排序，溢写到Linux磁盘。

那么现在就出现几个问题。

Q1：Map中的分区是按照什么来进行的？

A1：根据HashCode进行位运算然后取模得到。具体的可以看类的源码，也就是HashPartitioner。

Q2：怎么设置分区数量？

A2：这里的分区是跟后面的Reduce Task挂钩的，有多少个Reduce就会有几个分区。这里直接使用job的接口。job.setNumReduceTask(number)，number就是你想要的分区数。

Q3：分了区的后果是怎样的？

A3：首先有多少个Reduce就会生成多少个文件。如图：

很显然我们可以看到，第一张图只有一个大小为30B的文件，第二张图中有一个大小为7B的文件和23B的文件，之和也是30B，因为处理同样的数据，所以结果肯定是一样的。

其次，我们来看看如果分开两个文件他们的内容是什么。

很显然，如果把两个文件夹合起来就是用一个reduce输出的真正结果。也就是说，数据被分区处理，一个区的输出丢到一个Reduce中，生成一个文件。

Q4：怎么体现在分区的同时有排序的操作呢？

A4：还是看上面的两幅图。第一幅，开头首字母按照了字母便顺序来排序，第二个亦是如此。

Q5：为什么会溢写到磁盘？

A5：看图看图。

实际上这里存在着一个大小为100M的环形缓冲区，map不断向这里输入数据，当这里的资源占到了100M中的80%，就会把数据溢写到Linux磁盘。

再说Reuce Task

先看回到这两张图。

reduce进行的操作就是归并，聚集数据，把分散在各个地方的磁盘中的数据聚集到reduce，就是不断地拉数据。

这里在Map Task最边边那里可以看到“merge on disk”，也就是把相同分区地数据写到同一个磁盘上，然后Reduce Task的Shuffle就“fetch”拉数据，或者有可能不是本地的map，也要拉取Other maps。

注意：这里不同分区的数据是不同Reduce来拉的。也就是说，Reduce（1号）只会拉取一号分区的数据，不管是在本地还是在别的map上。

然后做合并就好了，这里一边合并，还会一边排序。

3

这里再介绍一个自定义的步骤，叫Combiner。

这是一个优化器，可以写也可以不写，在数据量很大的情况下可以有效地减少网络IO，减少Reduce压力！但是注意，这个Combiner是属于map中的Shuffle阶段！

例如同样是WordCount项目，hadoop又一百万个词条，hive有一千万个词条，那么怎么处理呢，这里就是运用Combiner进行一次提前的合并。使得传入Reduce的iter的参数：it（500,1000,600.....），而不是原来的 it（1,1,1,1,.....）。

这里有个问题，就是要区别Combiner和Reduce。

combiner对象是对于单个map来说的，只是处理单台机器生成的数据。

reduce对象是对于多个map来说的，所以如果有聚合，reduce阶段是不可避免的。

combiner获取的数据

<hadoop,list(1,1,1)>

combiner输出的数据

<hadoop,3>

打码

真的很开心