Hadoop学习（6）-- Hadoop MapReduce_执行任务过程中计算机宕机hadoop不会自动将计算机上执行的任务转移到其他计算

MapReduce设计构思

如何对付大数据处理场景

• 对相互间不具有计算依赖关系的大数据计算任务，实现并行最自然的办法就是采取MapReduce分而治之的策略。
• 首先Map阶段进行拆分，把大数据拆分成若干份小数据，多个程序同时并行计算产生中间结果；然后是Reduce聚合阶段，通过程序对并行的结果进行最终的汇总计算，得出最终的结果。
• 不可拆分的计算任务或相互间有依赖关系的数据无法进行并行计算！

构建抽象编程模型

• MapReduce借鉴了函数式语言中的思想，用Map和Reduce两个函数提供了高层的并行编程抽象模型。
  map: 对一组数据元素进行某种重复式的处理；
  reduce: 对Map的中间结果进行某种进一步的结果整理。
• MapReduce中定义了如下的Map和Reduce两个抽象的编程接口，由用户去编程实现:
  map: (k1; v1) → (k2; v2)
  reduce: (k2; [v2]) → (k3; v3)
• 通过以上两个编程接口，大家可以看出MapReduce处理的数据类型是<key,value>键值对。

统一架构、隐藏底层细节

• 如何提供统一的计算框架，如果没有统一封装底层细节，那么程序员则需要考虑诸如数据存储、划分、分发、结果收集、错误恢复等诸多细节；为此，MapReduce设计并提供了统一的计算框架，为程序员隐藏了绝大多数系统层面的处理细节。
• MapReduce最大的亮点在于通过抽象模型和计算框架把需要做什么(what need to do)与具体怎么做(how to do)分开了，为程序员提供一个抽象和高层的编程接口和框架。
• 程序员仅需要关心其应用层的具体计算问题，仅需编写少量的处理应用本身计算问题的业务程序代码。
• 至于如何具体完成这个并行计算任务所相关的诸多系统层细节被隐藏起来,交给计算框架去处理：从分布代码的执行，到大到数千小到单个节点集群的自动调度使用。

MapReduce介绍

MapReduce介绍

• Hadoop MapReduce是一个分布式计算框架，用于轻松编写分布式应用程序，这些应用程序以可靠，容错的方式并行处理大型硬件集群（数千个节点）上的大量数据（多TB数据集）。
• MapReduce是一种面向海量数据处理的一种指导思想，也是一种用于对大规模数据进行分布式计算的编程模型。

MapReduce特点

• 易于编程
  Mapreduce框架提供了用于二次开发的接口；简单地实现一些接口，就可以完成一个分布式程序。任务计算交给计算框架去处理，将分布式程序部署到hadoop集群上运行，集群节点可以扩展到成百上千个等。

• 良好的扩展性
  当计算机资源不能得到满足的时候，可以通过增加机器来扩展它的计算能力。基于MapReduce的分布式计算得特点可以随节点数目增长保持近似于线性的增长，这个特点是MapReduce处理海量数据的关键，通过将计算节点增至几百或者几千可以很容易地处理数百TB甚至PB级别的离线数据。

• 高容错性
  Hadoop集群是分布式搭建和部署得，任何单一机器节点宕机了，它可以把上面的计算任务转移到另一个节点上运行，不影响整个作业任务得完成，过程完全是由Hadoop内部完成的。

• 适合海量数据的离线处理
  可以处理GB、TB和PB级别得数据量。

MapReduce局限性

MapReduce虽然有很多的优势，也有相对得局限性，局限性不代表不能做，而是在有些场景下实现的效果比较差，并不适合用MapReduce来处理，主要表现在以下结果方面：

• 实时计算性能差
  MapReduce主要应用于离线作业，无法作到秒级或者是亚秒级得数据响应。
• 不能进行流式计算
  流式计算特点是数据是源源不断得计算，并且数据是动态的；而MapReduce作为一个离线计算框架，主要是针对静态数据集，数据是不能动态变化。
• 不擅长 DAG（有向无环图）计算
  多个应用程序存在依赖关系，后一个应用程序的输入为前一个的输出。在这种情况下，MapReduce 并不是不能做，而是使用后，每个 MapReduce 作业的输出结果都会写入到磁盘，会造成大量的磁盘 IO，导致性能非常的低下。

MapReduce实例进程

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

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

阶段组成

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

MapReduce数据类型

• 注意：整个MapReduce程序中，数据都是以kv键值对的形式流转的；
• 在实际编程解决各种业务问题中，需要考虑每个阶段的输入输出kv分别是什么；
• MapReduce内置了很多默认属性，比如排序、分组等，都和数据的k有关，所以说kv的类型数据确定及其重要的

MapReduce官方示例

wordcount单词词频统计

WordCount中文叫做单词统计、词频统计；指的是统计指定文件中，每个单词出现的总次数。

WordCount编程实现思路

• map阶段的核心：把输入的数据经过切割，全部标记1，因此输出就是<单词，1>。
• shuffle阶段核心：经过MR程序内部自带默认的排序分组等功能，把key相同的单词会作为一组数据构成新的kv对。
• reduce阶段核心：处理shuffle完的一组数据，该组数据就是该单词所有的键值对。对所有的1进行累加求和，就是单词的总次数。

WordCount执行流程图

Map阶段执行流程

MapReduce整体执行流程图

Map阶段执行过程

• 第一阶段：把输入目录下文件按照一定的标准逐个进行逻辑切片，形成切片规划。默认Split size = Block size（128M），每一个切片由一个MapTask处理。（getSplits）
• 第二阶段：对切片中的数据按照一定的规则读取解析返回<key,value>对。默认是按行读取数据。key是每一行的起始位置偏移量，value是本行的文本内容。（TextInputFormat）
• 第三阶段：调用Mapper类中的map方法处理数据。每读取解析出来的一个<key,value> ，调用一次map方法。
• 第四阶段：按照一定的规则对Map输出的键值对进行分区partition。默认不分区，因为只有一个reducetask。分区的数量就是reducetask运行的数量。
• 第五阶段：Map输出数据写入内存缓冲区，达到比例溢出到磁盘上。溢出spill的时候根据key进行排序sort。默认根据key字典序排序。
• 第六阶段：对所有溢出文件进行最终的merge合并，成为一个文件。

Reduce阶段执行流程

Reduce阶段执行过程

• 第一阶段：ReduceTask会主动从MapTask复制拉取属于需要自己处理的数据。
• 第二阶段：把拉取来数据，全部进行合并merge，即把分散的数据合并成一个大的数据。再对合并后的数据排序。
• 第三阶段：是对排序后的键值对调用reduce方法。键相等的键值对调用一次reduce方法。最后把这些输出的键值对写入到HDFS文件中。

Shuffle机制

shuffle概念

• Shuffle的本意是洗牌、混洗的意思，把一组有规则的数据尽量打乱成无规则的数据。
• 而在MapReduce中，Shuffle更像是洗牌的逆过程，指的是将map端的无规则输出按指定的规则“打乱”成具有一定规则的数据，以便reduce端接收处理。
• 一般把从Map产生输出开始到Reduce取得数据作为输入之前的过程称作shuffle。

Map端Shuffle

• Collect阶段：将MapTask的结果收集输出到默认大小为100M的环形缓冲区，保存之前会对key进行分区的计算，默认Hash分区。
• Spill阶段：当内存中的数据量达到一定的阀值的时候，就会将数据写入本地磁盘，在将数据写入磁盘之前需要对数据进行一次排序的操作，如果配置了combiner，还会将有相同分区号和key的数据进行排序。
• Merge阶段：把所有溢出的临时文件进行一次合并操作，以确保一个MapTask最终只产生一个中间数据文件。

Reducer端shuffle

• Copy阶段： ReduceTask启动Fetcher线程到已经完成MapTask的节点上复制一份属于自己的数据。
• Merge阶段：在ReduceTask远程复制数据的同时，会在后台开启两个线程对内存到本地的数据文件进行合并操作。
• Sort阶段：在对数据进行合并的同时，会进行排序操作，由于MapTask阶段已经对数据进行了局部的排序，ReduceTask只需保证Copy的数据的最终整体有效性即可。

shuffle机制弊端

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

Hadoop 序列化

序列化概述

什么是序列化

序列化就是把内存中的对象，转换成字节序列（或其他数据传输协议）以便于存储到磁
盘（持久化）和网络传输。
反序列化就是将收到字节序列（或其他数据传输协议）或者是磁盘的持久化数据，转换
成内存中的对象。

为什么要序列化

一般来说，“活的”对象只生存在内存里，关机断电就没有了。而且“活的”对象只能
由本地的进程使用，不能被发送到网络上的另外一台计算机。 然而序列化可以存储“活的”
对象，可以将“活的”对象发送到远程计算机。

为什么不用 Java 的序列化

Java 的序列化是一个重量级序列化框架（Serializable），一个对象被序列化后，会附带
很多额外的信息（各种校验信息，Header，继承体系等），不便于在网络中高效传输。所以，
Hadoop 自己开发了一套序列化机制（Writable）。

Hadoop 序列化特点：

（1）紧凑 ：高效使用存储空间。
（2）快速：读写数据的额外开销小。
（3）互操作：支持多语言的交互

自定义 bean 对象实现序列化接口（Writable）

在企业开发中往往常用的基本序列化类型不能满足所有需求，比如在 Hadoop 框架内部传递一个 bean 对象，那么该对象就需要实现序列化接口。
具体实现 bean 对象序列化步骤如下 7 步。

1. 必须实现 Writable 接口。

2. 反序列化时，需要反射调用空参构造函数，所以必须有空参构造。

   public FlowBean() {
   super();
   }
   1
   2
   3

3. 重写序列化方法

   @Override
   public void write(DataOutput out) throws IOException {
   out.writeLong(upFlow);
   out.writeLong(downFlow);
   out.writeLong(sumFlow);
   }
   1
   2
   3
   4
   5
   6

4. 重写反序列化方法

   @Override
   public void readFields(DataInput in) throws IOException {
   upFlow = in.readLong();
   downFlow = in.readLong();
   sumFlow = in.readLong();
   }
   1
   2
   3
   4
   5
   6

5. 注意反序列化的顺序和序列化的顺序完全一致。

6. 要想把结果显示在文件中，需要重写 toString()，可用"\t"分开，方便后续用。

7. 如果需要将自定义的 bean 放在 key 中传输，则还需要实现 Comparable 接口，因为
   MapReduce 框中的 Shuffle 过程要求对 key 必须能排序。

   @Override
   public int compareTo(FlowBean o) {
   // 倒序排列，从大到小
   return this.sumFlow > o.getSumFlow() ? -1 : 1;
   }
   1
   2
   3
   4
   5