Hadoop MapReduce

1.1 MapReduce介绍

MapReduce的思想核心是“先分再合，分而治之”。
所谓“分而治之”就是把一个复杂的问题，按照一定的“分解”方法分为等价的规模较小的若干部分，然后逐个解决，分别找出各部分的结果，然后把各部分的结果组成整个问题的最终结果。

• Map第一阶段，负责“拆分”：即把复杂的任务分解为若干个“简单的子任务”来并行处理。可以进行拆分的前提是这些小任务可以并行计算，彼此间几乎没有依赖关系
  拆分前提： 可并行计算+没有依赖关系
• Reduce第二阶段，负责“合并”：即对map阶段的结果进行全局汇总。
  
• MapReduce借鉴了函数式语言中的思想，用Map和Reduce两个函数提供了高层的并行编程抽象模型。
  map: 对一组数据元素进行某种重复式的处理；
  reduce: 对Map的中间结果进行某种进一步的结果整理
• MapReduce最大的亮点在于通过抽象模型和计算框架把需要做什么(业务问题)与具体怎么做(技术问题)分开了，为程序员提供一个抽象和高层的编程接口和框架。
  程序员仅需要关心其应用层的具体计算问题，仅需编写少量的处理应用本身计算问题的业务程序代码
• Hadoop MapReduce是一个分布式计算框架。
  分布式计算是一种计算方法，和集中式计算是相对的

1.2 MapReduce优缺点

优点：

• 易于编程：Mapreduce框架提供了用于二次开发的接口
• 良好的扩展性：当计算机资源不能得到满足的时候，可以通过增加机器来扩展它的计算能力。
• 高容错性：Hadoop集群是分布式搭建和部署得，任何单一机器节点宕机了，它可以把上面的计算任务转移到另一个节点上运行，不影响整个作业任务得完成
• 适合海量数据的离线处理：可以处理GB、TB和PB级别得数据量

局限性：MR主要是在离线计算领域

• 实时计算性能差。MapReduce主要应用于离线作业，无法作到秒级或者是亚秒级得数据响应
• 不能进行流式计算：流式计算特点是数据是源源不断得计算，并且数据是动态的；而MapReduce作为一个离线计算框架，主要是针对静态数据集得，数据是不能动态变化得

MapReduce实例进程

一个完整的MapReduce程序在分布式运行时有三类

1. MRAppMaster：负责整个MR程序的过程调度及状态协调
2. MapTask：负责map阶段的整个数据处理流程
3. ReduceTask：负责reduce阶段的整个数据处理流程

阶段组成

• 一个MapReduce编程模型中只能包含一个Map阶段和一个Reduce阶段，或者只有Map阶段；
• 不能有诸如多个map阶段、多个reduce阶段的情景出现；
• 如果用户的业务逻辑非常复杂，那就只能多个MapReduce程序串行运行

• 整个MapReduce程序中，数据都是以kv键值对的形式流转的

1.3 Hadoop MapReduce官方示例

• 一个最终完整版本的MR程序需要用户编写的代码和Hadoop自己实现的代码整合在一起才可以
• 由于MapReduce计算引擎天生的弊端（慢），当下企业中直接使用率已经很少了，所以在企业中工作很少涉及到MapReduce直接编程，但是某些软件的背后还依赖MapReduce引擎
• 但是后续的新的计算引擎比如Spark，当中就有MapReduce深深的影子存在

案例：评估圆周率π（PI）的值

蒙特卡洛方法计算，在平面上随机撒点

node1上：

jps  #验证Hadoop是否启动
start-all.sh #启动Hadoop

cd /export/server/hadoop-3.3.0/  #进入Hadoop安装包
cd share/
ll
cd hadoop/
ll
cd mapreduce/
ll
#可以看到一个jar文件
hadoop jar hadoop-mapreduce-examples-3.3.0.jar pi 2 2
#调用hadoop-mapreduce-examples-3.3.0.jar文件
#后面三个参数：pi表示MapReduce程序执行圆周率计算任务
#指定map阶段运行的任务task次数，并发度，这里是2；、
#每个map任务取样的个数，这里是2。

打开yarn页面：http://node1:8080/

案例：wordcount单词词频统计

统计文件中，每个单词出现的总次数
WordCount算是大数据计算领域经典的入门案例，相当于Hello World。

流程：

1. map阶段的核心：把输入的数据经过切割，全部标记1，因此输出就是<单词，1>。
   splite后进入map。因为MR数据类型都要求是keyvalue类型

2. shuffle阶段核心：经过MR程序内部自带默认的排序分组等功能，把key相同的单词会作为一组数据构成新的kv对
   根据key把他们分组，放在一起

3. reduce阶段核心：处理shuffle完的一组数据，该组数据就是该单词所有的键值对。对所有的1进行累加求和，就是单词的总次数

操作：

1. 准备数据：
   1.txt中存放要统计的内容
   打开node1:9870进入Hadoop （要先在node1上start-all.sh启动）
   在Hadoop上创建目录input，然后上传1.txt
2. 运行官方示例：
   官方实例位于Hadoop中mapReduce中

hadoop jar hadoop-mapreduce-examples-3.3.0.jar wordcount /input /outer
#依旧调用hadoop-mapreduce-examples-3.3.0.jar文件
#后面三个参数：wordcount表示执行单词统计任务；
#指定输入文件的路径
#指定输出结果的路径（该路径不能已存在）；

3. 查看结果
   打开hdfs，点进去ouput,有一个success是成功运行的标识文件，另一个文件显示输出结果
   

1.4 Map阶段执行流程

Map阶段执行过程：

1. 第一阶段：把输入目录下文件按照一定的标准逐个进行逻辑切片，形成切片规划。
   默认Split size = Block size（128M），每一个切片由一个MapTask处理。（getSplits）
   栗子：两个文件，文件a(300M)和文件b(200M)，需要3+2个切片，5个MapTask处理
2. 第二阶段：对切片中的数据按照一定的规则读取解析返回<key,value>对。
   默认是按行读取数据。key是每一行的起始位置偏移量，value是本行的文本内容。（TextInputFormat）
3. 第三阶段：调用Mapper类中的map方法处理数据。
4. 第四阶段：按照一定的规则对Map输出的键值对进行分区partition。默认不分区，因为只有一个reducetask。分区的数量就是reducetask运行的数量。
5. 第五阶段：Map输出数据写入内存缓冲区，达到比例溢出到磁盘上。溢出spill的时候根据key进行排序sort。默认根据key字典序排序。
   每次结果直接写入磁盘，io次数特别多，所以选择缓冲一下。类似水流打开冲击地面压力大，选择用一个杯子来缓冲，水杯接满一次倒地上一次，载接满再到地上
   缓冲区满了—即溢出spill
6. 第六阶段：对所有溢出文件进行最终的merge合并，成为一个文件。最后合并成一个文件
   
   input输入
   split切片，几个block数据块几个切片，
   memory buffer：缓冲区
   spill 溢写：同时sort排序
   merge合并成一个文件

1.5 Reduce阶段执行流程

1. 第一阶段：ReduceTask会主动从MapTask复制拉取属于需要自己处理的数据。
   map运行完后就把数据放在自己运行的本地，是reduce主动出击
2. 第二阶段：把拉取来数据，全部进行合并merge，即把分散的数据合并成一个大的数据。再对合并后的数据排序
   map阶段有多个maptask，数据从三个地方拉过来，所以需要合并
3. 第三阶段是对排序后的键值对调用reduce方法。键相等的键值对调用一次reduce方法。最后把这些输出的键值对写入到HDFS文件中。

copy — 合并排序 — 分组处理reduce

1.6 Shuffle机制

• Shuffle的本意是洗牌、混洗的意思，把一组有规则的数据尽量打乱成无规则的数据

• 而在MapReduce中，Shuffle更像是洗牌的逆过程，指的是将map端的无规则输出按指定的规则“打乱”成具有一定规则的数据，以便reduce端接收处理
  shuffle让数据有序

• 一般把从Map产生输出开始到Reduce取得数据作为输入之前的过程称作shuffle。处于下面红框中：
  

shuffle机制：是核心，但导致慢，慢的原因：数据在内存、磁盘之间的多次往复

1. Shuffle是MapReduce程序的核心与精髓，是MapReduce的灵魂
2. Shuffle也是MapReduce被诟病最多的地方所在。MapReduce相比较于Spark、Flink计算引擎慢的原因，跟Shuffle机制有很大的关系。
3. Shuffle中频繁涉及到数据在内存、磁盘之间的多次往复