一个完整的 mapreduce 程序在分布式运行时有三类实例进程：

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

map阶段：2步

　　第一步：设置InputFormat（通常使用TextInputFormat）的类型和数据的路径—获取数据的过程（可以得到K1,V1）

　　第二步：自定义Mapper–将K1，V1转为K2，V2

shuffle阶段：4步

　　第三步：分区的动作，如果有多个reduce才去考虑分区，默认只有一个reduce，分区可以省略

　　第四步：排序，默认对K2进行排序（字典序）–管好K2就行

　　第五步：规约，combiner是一个局部的reduce，map端的合并，是对mapreduce的优化操作，不会影响任何结果，减少网络传输，默认可以省略

　　第六步：分组，相同的K（K2）对应的V会放到同一个集合中—将map传递的K2，V2变成新的K2，V2

Reduce阶段：2步

　　第七步：自定义Reducer 得到K2，V2,转为K3，V3

　　第八步：设置OutputFormat和数据的路径–生成结果文件

序列化（Serialization）是指把结构化对象转化为字节流。

反序列化（Deserialization）是序列化的逆过程。把字节流转为结构化对象。

//反序列化的方法，反序列化时，
//从流中读取到的各个字段的顺序应该与序列化时
//写出去的顺序保持一致
 @Override 
 public void readFields(DataInput in) throws IOException { 
  upflow = in.readLong(); 
  dflow = in.readLong(); 
  sumflow = in.readLong(); 
 } 
 //序列化的方法 
 @Override 
 public void write(DataOutput out) throws IOException { 
  out.writeLong(upflow); 
  out.writeLong(dflow); 
  out.writeLong(sumflow); 
 } 
 @Override 
 public int compareTo(FlowBean o) { 
//实现按照 sumflow 的大小倒序排序 
  return sumflow>o.getSumflow()?-1:1; 
 } 

compareTo方法用于将当前对象与方法的参数进行比较。 如果指定的数与参数相等返回 0。 如果指定的数小于参数返回 -1。 如果指定的数大于参数返回 1。

例如：o1.compareTo(o2) 返回正数的话，当前对象（调用 compareTo 方法的对象 o1）要排在比较对象（compareTo 传参对象 o2）后面，返回负数的话，放在前面。

Mapreduce 中会将 map 输出的 kv 对，按照相同 key 分组，然后分发给不同的 reducetask。 默认的分发规则为：根据 key 的 hashcode%reducetask 数来分发

 

每一个 map 都可能会产生大量的本地输出，Combiner 的作用就是对 map 端的输出先做一次合并，以减少在 map 和 reduce 节点之间的数据传输量，以提高网络 IO 性能，是 MapReduce 的一种优化手段之一。

1. combiner 是 MR 程序中 Mapper 和 Reducer 之外的一种组件
2. combiner 组件的父类就是 Reducer
3. combiner 和 reducer 的区别在于运行的位置：
4. combiner 是在每一个 maptask 所在的节点运行
5. reducer 是接收全局所有 Mapper 的输出结果；
6. combiner 的意义就是对每一个 maptask 的输出进行局部汇总，以减小网络传输量

具体实现步骤：

1. 自定义一个 combiner 继承 Reducer，重写 reduce 方法
2. 在 job 中设置： job.setCombinerClass(CustomCombiner.class)

 

 

map 阶段处理的数据如何传递给 reduce 阶段，是 MapReduce 框架中最关键的一个流程，这个流程就叫 shuffle。

1. 将map端结果输出到环形缓冲区, 默认为100M, 保存的是<key, value>和分区信息多个ReduceTask时才需要分区
2. 当环形缓冲去到达80%时, 写入磁盘中, 在写入之前对数据进行快排, 如果配置了combiner, 还会对有相同分区号和key进行排序

　　　　Combiner规约：作用就是对map端的输出先做一次合并, 以减少在map和reduce节点之间的数据传输量, 以提高网络IO性能, 是MR的一种优化手段之一

3. 将所有溢出的临时文件进行一次合并操作, 确保一个MapTask最终只生成一个文件
4. Reduce复制一份数据, 默认保存在内存的缓冲区中, 到达阈值, 将数据写到磁盘
5. Reduce复制数据同时, 开启两个线程对内存到本地的数据进行合并
6. 进行合并数据同时, 进行排序, 因为Map端已进行局部排序, Reduce只需保证数据整体有效

　　　　压缩机制map输出压缩（减少shuffle过程中网络传输）reduce输出压缩（减少HDFS存储）

　　DEFLATE：自带，不可切分，和文本处理一样, 不需要修改

　　Gzip：自带，不可切分，和文本处理一样, 不需要修改

　　Bzip2：自带，可以切分，和文本处理一样, 不需要修改

　　LZO：需要安装，可以切分，建立索引指定输入格式

　　snappy：需要安装，不可切分，和文本处理一样, 不需要修改

 

1. 单纯按照文件内容切分
2. 切片大小等于block块大小
3. 不考虑数据集整体, 逐个针对文件切片
4. 切片公式: max(0, min(Long_max, blockSize))

 

　　Map 端 Join 解决数据倾斜，我们为每一个 MapTask 准备一个表的全表数据文件。这种机制叫做 Map Side Join。适用于关联表中有小表的情形。

CombineTextInputFormat

　　小文件处理场景 将 HDFS 上多个小文件合并到一个 InputSplit中，然后会启用一个 Map 来处理这里面的文件，以此减少 MR 整体作业的运行时间。

 

NameNode 维护整个文件系统元数据

按照类型划分：文件信息、块信息、datanode节点信息

按照形式划分：内存元数据、文件元数据

1. 内存元数据：文件信息、块信息、datanode节点信息
2. 文件元数据：只包含文件信息，其他信息是在datanode启动的时候进行上报

fsimage镜像文件

　　通常很大，保存的是一个小时前文件信息，是元数据的一个持久化的检查点，包含 Hadoop 文件系统中的所有目录和文件元数据信息，但不包含文件块位置的信息。文件块位置信息只存储在内存中，是在 datanode 加入集群的时候，namenode 询问 datanode 得到的，并且间断的更新。

edits日志

　　通常比较小，保存最近更改的文件信息，存放的是 Hadoop 文件系统的所有更改操作（文件创建，删除或修改）的日志，文件系统客户端执行的更改操作首先会被记录到 edits 文件中，只有当所有的写操作都执行完成之后，写操作才会返回成功。

SecondaryNameNode

　　职责是合并NameNode的edit logs到fsimage文件中, 减小edit logs文件的大小和得到一个最新的fsimage文件

Secondary Namenode将namenode上积累的所有edits和一个最新的fsimage下载到本地, 并加载到内存进行merge, 这个过程称为checkpoint. 一个小时检查一次fsimage，更新edits的数据到fsimage中。

 

HDFS所处的一种特殊状态，在这种状态下，文件系统只接受读数据请求，而不接受删除、修改等变更请求，是一种保护机制，用于保证集群中的数据块的安全性。 如果 HDFS 处于安全模式下，不允许 HDFS 客户端进行任何修改文件的操作,包括上传文件，删除文件，重命名，创建文件夹,修改副本数等操作。

 

是一个资源管理, 任务调度系统

　　ResourceManager：所有资源的监控, 分配和管理, 是一个全局的资源管理系统

　　NodeManager：每一个节点的维护，是每个节点上的资源和任务管理器，它是管理这台机器的代理，负责该节点程序的运行，以及该节点资源的管理和监控

　　ApplicationMaster：具体每一个应用程序的调度和协调, 分配任务, 用户提交的每个应用程序均包含一个AppMaster

对于所有的applications, RM拥有绝对的控制权和对资源的分配权. 而每个AM则会和RM协商资源, 同时和NodeManager通信来执行和监控task.

 

1. 客户端向RM提交程序, 包括AppMaster的必须信息，例如 ApplicationMaster 程序、启动 ApplicationMaster 的命令、用户程序等。
2. RM启动Container用来运行AppMaster
3. 启动中的AppMaster向RM注册, 启动完成后与RM保持心跳
4. AppMaster向RM申请相应数量的Container
5. RM返回AppMaster申请的Container信息, 由AppMaster对其进行初始化
6. AppMaster与NM进行通信, 要求NM启动Container, 两者保持心跳, 从而对NM运行的任务进行管理和监控
7. AppMaster对运行中的Container进行监控, Container通过RPC协议对相应的NM汇报信息
8. 客户端通过AppMaster获取运行状态进度等信息
9. 结束后, AppMaster向RM进行注销, 并回收Container

 

　　1）先进先出调度器(企业一定不会)：先进先出, 同一个时间队列只有一个任务执行

　　

　　2）容量调度器(Hadoop2.7.2默认调度器)：多队列, 每个队列内部先进先出, 同一个时间队列只有一个任务执行, 队列并行度为队列个数，

1. 1. 每个队列可配置资源
   2. 对同一用户提交的作业所占资源进行限定
   3. 首先, 计算每个队列中正在进行的任务与所分得资源的比值, 选出一个值最小的队列, 也就是最闲的
   4. 其次, 按照作业优先级和提交时间顺序, 同时考虑用户资源量限制和内存限制, 对队列中的任务排序
   5. 队列同时按照提交任务的顺序并行运行

　　　

 

HA(High Available), 高可用，是保证业务连续性的有效解决方案，一般有两个或两个以上的节点，分为活动节点（Active）及备用节点（Standby）。

 

在HA环境中, active的NM出现假死状态, stanby接收不到active的心跳, 判断active的NM处于宕机, 但实际上未死亡, stanby转换为active, 若此时原来active的NM复活, 此时有两台active的NM, 称为脑裂

危害

　　1. 造成资源争夺

　　2. 造成数据不统一

　　联邦机制就是会有多个NameNode。多个 NameNode 的情况意味着有多个 namespace(命名空间)，区别于 HA 模式下的多 NameNode，它们是拥有着同一个 namespace。

概括：

1. 多个 NN 共用一个集群里的存储资源，每个 NN 都可以单独对外提供服务。
2. 每个 NN 都会定义一个存储池，有单独的 id，每个 DN 都为所有存储池提供存储。
3. DN 会按照存储池 id 向其对应的 NN 汇报块信息，同时，DN 会向所有 NN 汇报本地存储可用资源情况。

缺点

　　并没有完全解决单点故障问题。 所以一般集群规模真的很大的时候，会采用 HA+Federation 的部署方案。也就是每个联合的 namenodes 都是 ha 的。

 

1. 在hdfs-site.xml文件中配置多目录，最好提前配置好，否则更改目录需要重新启动集群
2. NameNode有一个工作线程池，用来处理不同DataNode的并发心跳以及客户端并发的元数据操作。
3. dfs.namenode.handler.count=20 * log2(Cluster Size)，比如集群规模为10台时，此参数设置为60
4. 编辑日志存储路径dfs.namenode.edits.dir设置与镜像文件存储路径dfs.namenode.name.dir尽量分开，达到最低写入延迟
5. 服务器节点上YARN可使用的物理内存总量，默认是8192（MB），注意，如果你的节点内存资源不够8GB，则需要调减小这个值，而YARN不会智能的探测节点的物理内存总量。　　yarn.nodemanager.resource.memory-mb
6. 单个任务可申请的最多物理内存量，默认是8192（MB）。yarn.scheduler.maximum-allocation-mb

 

Hadoop宕机

1. 如果MR造成系统宕机。此时要控制Yarn同时运行的任务数，和每个任务申请的最大内存。调整参数：yarn.scheduler.maximum-allocation-mb（单个任务可申请的最多物理内存量，默认是8192MB）
2. 如果写入文件过量造成NameNode宕机。那么调高Kafka的存储大小，控制从Kafka到HDFS的写入速度。高峰期的时候用Kafka进行缓存，高峰期过去数据同步会自动跟上。

 

Hadoop基准测试