MapReduce 机理

1.hadoop 平台进程

Namenode进程:

管理者文件系统的Namespace。它维护着文件系统树(filesystem tree)以及文件树中所有的文件和文件夹的元数据(metadata)。管理这些信息的文件有两个，分别是Namespace 镜像文件(Namespace image)和操作日志文件(edit log)，这些信息被Cache在RAM中，当然，这两个文件也会被持久化存储在本地硬盘。Namenode记录着每个文件中各个块所在的数据节点的位置信息，但是他并不持久化存储这些信息，因为这些信息会在系统启动时从数据节点重建。

Namenode容错机制

第一种方式是将持久化存储在本地硬盘的文件系统元数据备份。Hadoop可以通过配置来让Namenode将他的持久化状态文件写到不同的文件系统中。这种写操作是同步并且是原子化的。比较常见的配置是在将持久化状态写到本地硬盘的同时，也写入到一个远程挂载的网络文件系统。

第二种方式是运行一个辅助的Namenode(Secondary Namenode)。事实上Secondary Namenode并不能被用作Namenode它的主要作用是定期的将Namespace镜像与操作日志文件(edit log)合并，以防止操作日志文件(edit log)变得过大。通常，Secondary Namenode 运行在一个单独的物理机上，因为合并操作需要占用大量的CPU时间以及和Namenode相当的内存。辅助Namenode保存着合并后的Namespace镜像的一个备份，万一哪天Namenode宕机了，这个备份就可以用上了。

但是辅助Namenode总是落后于主Namenode，所以在Namenode宕机时，数据丢失是不可避免的。在这种情况下，一般的，要结合第一种方式中提到的远程挂载的网络文件系统(NFS)中的Namenode的元数据文件来使用，把NFS中的Namenode元数据文件，拷贝到辅助Namenode，并把辅助Namenode作为主Namenode来运行。

DATANODE

Datanode是文件系统的工作节点，他们根据客户端或者是namenode的调度存储和检索数据，并且定期向namenode发送他们所存储的块(block)的列表。

集群中的每个服务器都运行一个DataNode后台程序，这个后台程序负责把HDFS数据块读写到本地的文件系统。当需要通过客户端读/写某个数据时，先由NameNode告诉客户端去哪个DataNode进行具体的读/写操作，然后，客户端直接与这个DataNode服务器上的后台程序进行通信，并且对相关的数据块进行读/写操作。

Secondary NameNode

Secondary NameNode不同于NameNode，它不接受或者记录任何实时的数据变化，但是，它会与NameNode进行通信，以便定期地保存HDFS元数据的快照。由于NameNode是单点的，通过Secondary NameNode的快照功能，可以将NameNode的宕机时间和数据损失降低到最小。同时，如果NameNode发生问题，Secondary NameNode可以及时地作为备用NameNode使用。

（1） Secondary NameNode请求NameNode进行edit log的滚动（即创建一个新的edit log），将新的编辑操作记录到新生成的edit log文件；

（2）通过http get方式，读取NameNode上的fsimage和edits文件，到Secondary NameNode上；

（3）读取fsimage到内存中，即加载fsimage到内存，然后执行edits中所有操作（类似OracleDG，应用redo log），并生成一个新的fsimage文件，即这个检查点被创建；

（4）通过http post方式，将新的fsimage文件传送到NameNode；

（5） NameNode使用新的fsimage替换原来的fsimage文件，让（1）创建的edits替代原来的edits文件；并且更新fsimage文件的检查点时间。

ResourceManager（RM）
ResourceManage 即资源管理，在YARN中，ResourceManager负责集群中所有资源的统一管理和分配，它接收来自各个节点（NodeManager）的资源汇报信息，并把这些信息按照一定的策略分配给各个应用程序（实际上是ApplicationManager）。

资源分配单位用“资源容器”（Contrainer）表示，Container是一个动态资源分配单位，它将内存、cpu、磁盘、网络等资源封装在一起，从而限定每个任务使用的资源量。

1）处理客户端请求；
2）启动或监控ApplicationMaster；
3）监控NodeManager；
4）资源的分配与调度。

NodeManager（NM）

NodeManager进程运行在集群中的节点上，每个节点都会有自己的NodeManager。Nodemanager整个集群有多个，负责每个节点上的资源和使用。NodeManager是一个slave服务：

它负责接收处理来自ResourceManager的资源分配请求，分配具体的Container给应用。
同时，它还负责监控并报告Container使用信息给ResourceManager。NodeManager只负责管理自身的Container，它并不知道运行在它上面应用的信息。负责管理应用信息的组件是ApplicationMaster
NodeManager的任务包括：

1）和ResourceManager保持同步
2）跟踪Node的状态
3）监控Container的生命周期，监控Container使用的资源
4）管理Distributed Cache
5）管理Container生成的日志

MapReduce机理

1.1 MapTask
工作机制简述：输入的数据通过 split 被逻辑切分为多个 split 文件，通过 Record 按行读取内容给 map（用户实现）进行处理，数据被 map 处理结束后交给 OutputCollector 收集器，对其结果 key 进行分区（hash），然后写入 buffer，每个 maptask 都有一个内存缓冲区，存储着 map 的输出结果，当缓冲区快满时需要将缓冲区的数据以一个临时文件的方式存放到磁盘，当整个 maptask 结束后，再对磁盘中 maptask 产生的所有临时文件做合并，生成最终的正式输出文件，然后等待 reducetask 获取该数据。

1.2 ReduceTask
工作机制简述： Reduce 分为 copy、sort、reduce 三个阶段。ReduceTask 从各个 MapTask 上远程 copy 一片数据，进行一次归并排序，最后由 reduce（用户实现）将数据写到 HDFS 上。

>>>>>2. 数据输入
数据输入：输入文件的格式包括：日志文件、二进制文件、数据库表等，具体的文件类型，就需要对应的 InputFormat 来读取数据。

常用的 InputFormat 的实现类：

FileInputFormat：按照文件的内容长度进行切片，切片大小默认等于 Block 大小，切片时不需要考虑数据集整体，只是针对每一个文件单独切片。
TextInputFormat：是 FileInputFormat 默认的实现类，按照行读取每行记录，键是存储改行在整个文件中的起始字节偏移量（offset），LongWritable 类型；值是该行的内容，Text 类型。
CombineTextInputFormat：用于小文件过多的场景，它可以将多个小文件从逻辑上规划到一个切片中，及将多个小文件交给一个 MapTask 处理。

数据块： 是 HDFS 存储的数据单位，实际上是 HDFS 将文件分成一块一块（Block）

数据切片： 是 MapReduce 程序计算输入数据的单位，一个切片会启动一个对应的 MapTask（数据切片只是逻辑上对输入的数据进行分片）

MapTask 原理

inputformat 阶段： inputFormat（默认 TextInputFormat）通过 getSplits 方法对输入目录中的文件进行逻辑切片，得到 block，但后分配对应数量的 MapTask。

read 阶段： MapTask 通过 inputFormat 获取 recordReader 对象（默认 LineRecordReader）进行读取，以 \n 最为分隔符，读取一行数据，返回 <key, value>。key 表示每行首字符的字节偏移量，value 表示这一行的内容

map 阶段： 将解析出的 <key, value> 交给用户自己继承的 Mapper 类中，执行其中的 map 方法。RecordReader 每读取一行，在这里调用一次！

Collect 阶段： Mapper 的逻辑结束后，将每条数据通过 contect.write 进行收集数据，通过 OutputCollect.collect() 输出结果；在 collect() 方法中，会先对其进行分区（默认使用 HashPartitioner）处理，并写入一个环形缓冲区中

MapTask 提供的 Partitioner 接口，其作用是根据 key 或 value 及 reducer 的数量来决定当前输出数据应该交由 reducetask 处理，默认对 key hash 后以 reducer 数量取模（默认的取模方式是平均 reducer 的处理能力）

缓冲区的作用是批量收集 Mapper 结果，减少磁盘 I/O 的影响，key/value 对以及 partition 的结果都会被写入缓冲区（key、value 都会被序列化成字节数组）。

Spill 阶段： 当溢写线程启动后，对着缓冲区80%空间内的 key 做依次本地排序（sort），并在必要时对数据进行合并、压缩等操作

溢写的流程

1、利用快速排序算法对缓冲区内的数据进行排序，先按照分区编号排序，然后每个分区内按照key排序。

2、按照分区编号将每个分区中的数据写入任务工作目录下的临时文件 output/spillN.out（N表示当前溢写的次数）。如果有 Combiner ，则写入文件之前，对每个分区内的数据进行依次聚集操作。

3、将分区数据的元数据写到内存索引数据结构 SpillRecord 中。如果当前内存索引大小超过1MB，则将内存索引写到文件 output/spillN.out.index 中。

Merge 阶段：每次溢写会在磁盘上生成一个临时文件（写之前判断是否需有 Combiner），当数据处理结束后，会对所有的临时文件进行一次合并，确保最终只生成一个文件（output/file.out），同时生成一个索引文件（output/file.out.index）以记录数据偏移量。

环形缓冲区

环形缓冲区是一个数组，数组中存放着 key、value 的序列化数据以及相关的元数据，包括 partition、key的起始位置、value的起始位置、value的长度等。

缓冲区的默认大小是100MB，当缓冲区80%的空间被占用是，就会启动溢写（Spill），Mapper 的输出结果可以往剩下20%的空间继续写入数据。

MapTask 并行度决定依据

一个 Job 的 Map 阶段并行度由 Job 的分片数决定。每一个 Split 分片分配一个 MapTask 进行处理。

Shuffle

Shuffle： map 阶段完成后，数据需要传输到 reduce，这中间的（数据处理）过程可称为 shuffle（洗牌，发牌）。

shuffle的核心机制包括：数据分区、排序、分组、规约、合并等过程。

input Split： 输入的数据分片后，每一个 split 都会由一个 Mapreduce 处理，最先由 map 进行处理

Map 阶段的 Shuffle：

Collect 阶段： 将MapTask的结果输出到默认大小为100M的环形缓冲区，保存的是key/value序列化数据，Partition分区信息等。

Spill 阶段： 当内存中的数据量达到一定的阀值（80%）的时候，就会将数据写入本地磁盘。在将数据写入磁盘之前，线程首先根据数据最终要传递到的 Reduce 任务把数据划分成相应的分区（Partition），每个分区中对数据的 key 进行一次排序的操作，如果配置了combiner（预聚合），还会将有相同分区号和 key 的数据进行聚合。

Merge 阶段： 把所有溢出的临时文件进行一次合并操作，以确保一个MapTask最终只产生一个中间数据文件。

溢出写文件归并完毕后，Map 任务将删除所有的临时溢出写文件，并告知 TaskTracker 任务已完成，只要其中一个 Map 任务完成，Reduce 任务就会开始复制它的输出（Copy 阶段)。

Map 任务的输出文件放置在运行 Map 任务的 TaskTracker 的本地磁盘上，它是运行 Reduce 任务的 TaskTracker 所需要的输入数据。

Reduce 阶段的 Shuffle：

Copy 阶段：ReduceTask启动Fetcher线程到已经完成MapTask的节点上，根据自己的分区号，复制一份属于自己的数据，这些数据默认会保存在内存的缓冲区中，当内存的缓冲区达到一定的阀值的时候，就会将数据写到磁盘之上。

Merge 阶段：在ReduceTask远程复制数据的同时，会在后台开启两个线程(一个是内存到磁盘的合并，一个是磁盘到磁盘的合并)，对本地的数据文件进行合并操作。

merge 有三种形式：

内存到内存：该形式不启用

内存到磁盘：该形式一直在运行

磁盘到磁盘：marge阶段，生成最终的文件

Sort 阶段：在对数据进行合并的同时，会进行排序操作，由于MapTask 阶段已经对数据进行了局部的排序，ReduceTask只需对Copy的数据进行归并排序。

最终文件默认位于磁盘中，当 Reduce 的输出文件已定，整个 Shuffle 阶段就结束了，但后进行 Reducer 的执行（从文件中取出每一个键值对，调用用户自定义的 reduce 方法）。

Partition 分区

当数据文件过大时，我们可以按照不同的条件将大文件拆分为小分进行处理，比如按照区域划分、按照日期划分等。

分区： MapReduce 在 Map 阶段结束后，会对数据按照 key 进行分区。

默认分区数据量为1，如果job中设置了 reduceTask 数量，则分区数量与 reduceTask 数量一致。

Partitioner

在 Driver 中，我们可以通过 job.setNumReduceTasks(Num) 来自定义分区数量，在理解之前就先试下。

对原本输出一个文件的案例的 Driver 修改其 ReduceTasks 数量：

分区数量

分区数 = reduce数：理想状态

分区数 > reduce数：任务报错

分区数 < reduce数：生成多余空文件

Comparable 排序

MapTask 和 ReduceTask 均会对数据按照 key 的字典顺序进行（快速）排序。该操作属于Hadoop的默认行为。任何应用程序中的数据均会被排序，而不管逻辑上是否需要。

Task的排序
对于 MapTask ，它会将处理的结果暂时放到环形缓冲区中，当环形缓冲区使用率达到一定阈值后，再对缓冲区中的数据进行一次快速排序，并将这些有序数据溢写到磁盘上，而当数据处理完毕后，它会对磁盘上所有文件进行归并排序。

对于 ReduceTask ，它从每个 MapTask 上远程拷贝相应的数据文件，如果文件大小超过一定阈值，则溢写磁盘上，否则存储在内存中。如果磁盘上文件数目达到一定阈值，则进行一次归并排序以生成一个更大文件；如果内存中文件大小或者数目超过一定阈值，则进行一次合并后将数据溢写到磁盘上。当所有数据拷贝完毕后，ReduceTask 统一对内存和磁盘上的所有数据进行一次归并排序。

排序分类
部分排序： MapReduce根据输入记录的键对数据集排序。保证输出的每个文件内部有序。

全局排序：最终输出结果只有一个文件，且文件内部有序。实现方式是只设置一个ReduceTask。但该方法在处理大型文件时效率极低，因为一台机器处理所有文件，完全丧失了MapReduce所提供的并行架构。

辅助排序：（GroupingComparator分组），在Reduce端对key进行分组。应用于：在接收的key为bean对象时，想让一个或几个字段相同（全部字段比较不相同）的key进入到同一个reduce方法时，可以采用分组排序。

⼆次排序：如果需要中间过程对key的分组规则和reduce前对key的分组规则不同，那么可以通过 JobConf.setOutputValueGroupingComparator(Class)来指定一个Comparator。再加上 JobConf.setOutputKeyComparatorClass(Class) 可用于控制中间过程的key如何被分组，所以结合两者可以实现按值的二次排序。