大数据开发工程师-第五周 第一章 初识MapReduce

---

第五周 第一章 初识MapReduce

Hadoop之MapReduce详解

前面我们学习了Hadoop中的HDFS，HDFS主要是负责存储海量数据的，如果只是把数据存储起来，除了浪费磁盘空间，是没有任何意义的，我们把数据存储起来之后是希望能从这些海量数据中分析出来一些有价值的内容，这个时候就需要有一个比较厉害的计算框架，来快速计算这一批海量数据，所以MapReduce应运而生了，那MapReduce是如何实现对海量的快速计算的呢？它的底层原理是什么样的呢？不要着急，且听下面分解。

MapReduce介绍

在这里我们先举个例子来介绍一下MapReduce
计算扑克牌中的黑桃个数
就是我们平时打牌时用的扑克牌，现在呢，有一摞牌，我想知道这摞牌中有多少张黑桃
最直接的方式是一张一张检查并且统计出有多少张是黑桃，但是这种方式的效率比较低，如果说这一摞牌
只有几十张也就无所谓了，如果这一摞拍有上千张呢？你一张一张去检查还不疯了？
这个时候我们可以使用MapReduce的计算方法
第一步：把这摞牌分配给在座的所有玩家
第二步：让每个玩家查一下自己手中的牌有多少张是黑桃，然后把这个数目汇报给你
第三步：你把所有玩家告诉你的数字加起来，得到最终的结果
之前是一张一张的串行计算，现在使用mapreduce是把数据分配给多个人，并行计算，每一个人获得一
个局部聚合的临时结果，最终再统一汇总一下。

这样就可以快速得到答案了，这其实就是MapReduce的计算思想。
下面我们再通过具体的案例分析MapReduce的计算思想

分布式计算介绍

移动数据计算(传统计算)

再举一个例子，就拿我们平时使用比较多的JDBC代码执行的流程来说。
我们自己写的JDBC代码是在一台机器上运行，mysql数据库是在另一台机器上运行。
正常情况下，我们通过jdbc代码去mysql中获取一条数据，速度还是很快的，但是有一个极端情况，如果我们要获取的数据量很大，达到了几个G，甚至于几十G。
这个时候我们使用jdbc代码去拉取数据的时候，就会变得非常慢，这个慢主要是由于两个方面造成的，
一个是磁盘io(会进行磁盘读写操作)，
一个是网络io(网络传输)。 这两个里面其实最耗时的还是网络io，我们平时在两台电脑之间传输一个几十G的文件也需要很长时间的，但是如果是使用U盘拷贝就很快了，所以可以看出来主要耗时的地方是在网络IO上面。

这种计算方式我们称之为移动数据 ，就是把mysql数据库中的数据移动到计算程序所在的机器上面

如果我们考虑把计算程序移动到mysql上面去执行，是不是就可以节省网络io了，是的！
这种方式称之为移动计算，就是把计算程序移动到数据所在的节点上面

移动计算程序计算(新思路)

如果我们数据量很大的话，我们的数据肯定是由很多个节点存储的，这个时候我们就可以把我们的程序代码拷贝到对应的节点上面去执行，程序代码都是很小的，一般也就几十KB或者几百KB，加上外部依赖 包，最大也就几兆 ，甚至几十兆，但是我们需要计算的数据动辄都是几十G、几百G，他们两个之间的差距不是一星半点啊
这样我们的代码就可以在每个数据节点上面执行了，但是这个代码只能计算当前节点上的数据的，如果我们想要统计数据的总行数，这里每个数据节点上的代码只能计算当前节点上数据的行数，所以还的有一个汇总程序，这样每个数据节点上面计算的临时结果就可以通过汇总程序得到最终的结果了。
此时汇总程序需要传递的数据量就很小了，只需要接收一个数字即可。

这个计算过程就是分布式计算，这个步骤分为两步:
第一步：对每个节点上面的数据进行局部计算
第二步：对每个节点上面计算的局部结果进行最终全局汇总

MapReduce原理剖析

MapReduce是一种分布式计算模型，是Google提出来的，主要用于搜索领域，解决海量数据的计算问题.
MapReduce是分布式运行的，由两个阶段组成：Map和Reduce，Map阶段是一个独立的程序，在很多个节点同时运行，每个节点处理一部分数据。
Reduce阶段也是一个独立的程序，可以在一个或者多个节点同时运行，每个节点处理一部分数据【在这我们先把reduce理解为一个单独的聚合程序即可】。在这map就是对数据进行局部汇总，reduce就是对局部数据进行最终汇总。

结合到我们前面分析的统计黑桃的例子中，这里的map阶段就是指每个人统计自己手里的黑桃的个数，reduce就是对每个人统计的黑桃个数进行最终汇总

这是一个Hadoop集群，一共5个节点一个主节点，四个从节点
这里面我们只列出来了HDFS相关的进程信息
假设我们有一个512M的文件，这个文件会产生4个block块，假设这4个block块正好分别存储到了集群的4个节点上，我们的计算程序会被分发到每一个数据所在的节点，然后开始执行计算，在map阶段，针对每一个block块对应的数据都会产生一个map任务(这个map任务其实就是执行这个计算程序的)，在这里也就意味着会产生4个map任务并行执行，4个map阶段都执行完毕以后，会执行reduce阶段，在reduce阶段中会对这4个map任务的输出数据进行汇总统计，得到最终的结果

  左下角是一个文件，文件最下面是几个block块，说明这个文件被切分成了这几个block块，文件上面是一些split，注意，咱们前面说的每个block产生一个map任务，其实这是不严谨的，其实严谨一点来说的话应该是一个split产生一个map任务。
  那这里的block和split之间有什么关系吗？ 我们来分析一下
block块是文件的物理切分，在磁盘上是真实存在的。是对文件的真正切分而split是逻辑划分，不是对文件真正的切分，默认情况下我们可以认为一个split的大小和一个block的大小是一样的，所以实际上是一个split会产生一个map task
  这里面的map Task就是咱们前面说的map任务，看后面有一个reduce Task，reduce会把结果数据输出到hdfs上，有几个reduce任务就会产生几个文件，这里有三个reduce任务，就产生了3个文件，咱们前面分析的案例中只有一个reduce任务做全局汇总
  注意看map的输入输出 reduce的输入输出map的输入是k1,v1 输出是k2,v2 reduce的输入是k2,v2 输出是k3,v3 都是键值对的形式。
  在这注意一下，为什么在这是1,2,3呢？ 这个主要是为了区分数据，方便理解，没有其它含义，这是我们人为定义的

MapReduce之Map阶段(六步)

  mapreduce主要分为两大步骤 map和reduce，map和reduce在代码层面对应的就是两个类，map对应的是mapper类，reduce对应的是reducer类，下面我们就来根据一个案例具体分析一下这两个步骤
假设我们有一个文件，文件里面有两行内容
第一行是hello you
第二行是hello me
我们想统计文件中每个单词出现的总次数

首先是map阶段
第一步：框架会把输入文件(夹)划分为很多Input Split，这里的inputsplit就是前面我们所说的split【对文件进行逻辑划分产生的】，默认情况下，每个HDFS的Block对应一个Input Split。再通过RecordReader类，把每个InputSplit解析成一个一个的<k1,v1>。默认情况下，每一行数据，都会被解析成一个<k1,v1>
这里的k1是指每一行的起始偏移量，v1代表的是那一行内容，所以，针对文件中的数据，经过map处理之后的结果是这样的
<0，hello you>
<10，hello me>
注意：map第一次执行会产生<0，hello you>，第二次执行会产生<10，hello me>，并不是执行一次就获取到这两行结果了，因为每次只会读取一行数据，我在这里只是把这两行执行的最终结果都列出来了

第二步：框架调用Mapper类中的map(…)函数，map函数的输入是<k1,v1>，输出是<k2,v2>。一个InputSplit对应一个map task。程序员需要自己覆盖Mapper类中的map函数，实现具体的业务逻辑。
因为我们需要统计文件中每个单词出现的总次数，所以需要先把每一行内容中的单词切开，然后记录出现次数为1,这个逻辑就需要我们在map函数中实现了
那针对<0，hello you>执行这个逻辑之后的结果就是
<hello,1>
<you,1>
针对<10，hello me>执行这个逻辑之后的结果是
<hello,1>
<me,1>

第三步：框架对map函数输出的<k2,v2>进行分区。不同分区中的<k2,v2>由不同的reduce task处理，默认只有1个分区，所以所有的数据都在一个分区，最后只会产生一个reduce task。
经过这个步骤之后，数据没什么变化，如果有多个分区的话，需要把这些数据根据分区规则分开，在这里默认只有1个分区。
<hello,1>
<you,1>
<hello,1>
<me,1>
咱们在这所说的单词计数，其实就是把每个单词出现的次数进行汇总即可，需要进行全局的汇总，不需要进行分区，所以一个redeuce任务就可以搞定，如果你的业务逻辑比较复杂，需要进行分区，那么就会产生多个reduce任务了，那么这个时候，map任务输出的数据到底给哪个reduce使用？这个就需要划分一下，要不然就乱套了。
假设有两个reduce，map的输出到底给哪个reduce，如何分配，这是一个问题。
这个问题，由分区来完成。map输出的那些数据到底给哪个reduce使用，这个就是分区干的事了。

第四步：框架对每个分区中的数据，都会按照k2进行排序、分组。分组指的是相同k2的v2分成一个组。
先按照k2排序
<hello,1>
<hello,1>
<me,1>
<you,1>
然后按照k2进行分组，把相同k2的v2分成一个组
<hello,{1,1}>
<me,{1}>
<you,{1}>

第五步：在map阶段，框架可以选择执行Combiner过程
Combiner可以翻译为规约，规约是什么意思呢？ 在刚才的例子中，咱们最终是要在reduce端计算单词 出现的总次数的，所以其实是可以在map端提前执行reduce的计算逻辑，先对在map端对单词出现的次数进行局部求和操作，这样就可以减少map端到reduce端数据传输的大小，这就是规约的好处，当然了，并不是所有场景都可以使用规约，针对求平均值之类的操作就不能使用规约了，否则最终计算的结果就不准确了。Combiner一个可选步骤，默认这个步骤是不执行的。

第六步：框架会把map task输出的<k2,v2>写入到linux 的磁盘文件中
<hello,{1,1}>
<me,{1}>
<you,{1}>
至此，整个map阶段执行结束

最后注意一点：
MapReduce程序是由map和reduce这两个阶段组成的，但是reduce阶段不是必须的，也就是说有的mapreduce任务只有map阶段，为什么会有这种任务呢？
  是这样的，咱们前面说过，其实reduce主要是做最终聚合的，如果我们这个需求是不需要聚合操作，直接对数据做过滤处理就行了，那也就意味着数据经过map阶段处理完就结束了，所以如果reduce阶段不存在的话，map的结果是可以直接保存到HDFS中的
注意，如果没有reduce阶段，其实map阶段只需要执行到第二步就可以，第二步执行完成以后，结果就可以直接输出到HDFS了。
针对我们这个单词计数的需求是存在reduce阶段的，所以我们继续往下面分析。

MapReduce之Reduce阶段(十步)

第一步：框架对多个map任务的输出，按照不同的分区，通过网络copy到不同的reduce节点。这个过程称作shuffle
针对我们这个需求，只有一个分区，所以把数据拷贝到reduce端之后还是老样子
<hello,{1,1}>
<me,{1}>
<you,{1}>
第二步：框架对reduce端接收的相同分区的<k2,v2>数据进行合并、排序、分组。
reduce端接收到的是多个map的输出，对多个map任务中相同分区的数据进行合并 排序 分组
注意，之前在map中已经做了排序 分组，这边也做这些操作 重复吗？
不重复，因为map端是局部的操作 reduce端是全局的操作
之前是每个map任务内进行排序，是有序的，但是多个map任务之间就是无序的了。
不过针对我们这个需求只有一个map任务一个分区，所以最终的结果还是老样子
<hello,{1,1}>
<me,{1}>
<you,{1}>
第三步：框架调用Reducer类中的reduce方法，reduce方法的输入是<k2,{v2}>，输出是<k3,v3>。一个<k2,{v2}>调用一次reduce函数。程序员需要覆盖reduce函数，实现具体的业务逻辑。
那我们在这里就需要在reduce函数中实现最终的聚合计算操作了，将相同k2的{v2}累加求和，然后再转化为k3,v3写出去，在这里最终会调用三次reduce函数
<hello,2>
<me,1>
<you,1>
第四步：框架把reduce的输出结果保存到HDFS中。
hello 2
me 1
you 1
至此，整个reduce阶段结束。

wordcount实例简略过程

map阶段:
第一步:每个inputsplit，用recordreader类划分成<k1,v1>
<0，hello you>
<10，hello me>
第二步:调用map()函数，得到<k2,v2>
<hello,1>
<you,1>
<hello,1>
<me,1>
第三步:分区
<hello,1>
<you,1>
<hello,1>
<me,1>
第四步:排序分组
<hello,{1,1}>
<you,1>
<me,1>
第五步:combiner规约(可以无)

reduce阶段
第一步:
根据分区的不同将<k2,{v2,v2...}> copy到不同的reduce task
<k2,{v2,v2...}>
第二步:
对同一分区<k2,{v2,v2...}>进行合并排序分组，得<k2,{v2,v2...}>
第三步:
每个<k2,{v2,v2...}>调用reducer类的reduce方法，得<k3,v3>
第四步:
第四步框架把reduce的输出结果保存到HDFS中。
hello 2
me 1
you 1

---

: