简介
主要记录了Hadoop各个组件的基本原理，处理过程和关键的知识点等，包括HDFS、YARN、MapReduce等。
铺垫
- 人产生数据的速度越来越快，机器则更加快，数据的增长速度通常比算法更快，所以需要另外的一种处理数据的方法。
- 硬盘的容量增加了，但性能没有跟上，解决办法是把数据分到多块硬盘，然后同时读取。但带来一些问题：
硬件问题：复制数据解决（RAID）
分析需要从不同的硬盘读取数据：MapReduce
而Hadoop提供了
1.可靠的共享存储（分布式存储）
2.抽象的分析接口（分布式分析）
大数据
概念
不能使用一台机器 进行处理的数据
大数据的核心是样本 = 总体
特性
- 大量性（volume）：一般在大数据里，单个文件的级别至少为几十，几百GB以上
- 快速性（velocity）：反映在数据快速产生以及数据变更的频率上
- 多样性（variety）：泛指数据类型及其来源的多样性，进一步可以把数据结构归纳为结构化(structured)，半结构化(semi-structured)，和非结构化(unstructured)
- 易变性：伴随数据快速性的特征，数据流还呈现出一种波动的特征。不稳定的数据会随着日，季节，特定事件的触发出现周期性峰值
- 准确性：又称为数据保证（data assurance），不同方式，收集到的数据会在质量上很大差异，数据分析和输出结果的错误程度和可信度在很大程度上取决于收集到的数据质量的高低
- 复杂性：体现在数据的管理和操作上。如何抽取，转换，加载，连接，关联以把握数据内蕴的有用信息已经变得越来越有挑战性
关键技术
1.数据分布在多台机器
可靠性：每个数据块都复制到多个节点
性能：多个节点同时处理数据
2.计算随数据走
网络IO速度 << 本地磁盘IO速度，大数据系统会尽量地将任务分配到离数据最近的机器上运行（程序运行时，将程序及其依赖包都复制到数据所在的机器运行）
代码向数据迁移，避免大规模数据时，造成大量数据迁移的情况，尽量让一段数据的计算发生在同一台机器上
3.串行IO取代随机IO
传输时间 << 寻道时间，一般数据写入后不再修改
Hadoop简介
Hadoop可运行在 一般的商用服务器上，具有高容错，高可靠性，高扩展性等特点
特别适合 写一次 读多次的场景
适合
- 大规模数据
- 流式数据（写一次，读多次）
- 商用硬件（一般硬件）
不适合
- 低延时的数据访问
- 大量的小文件
- 频繁修改文件（基本就是写1次）
Hadoop架构
- HDFS：分布式文件存储
- YARN：分布式资源管理
- MapReduce：分布式计算
- Others：利用YARN的资源管理功能能实现其他的数据处理方式
内部各个节点基本都是采用Master-Woker架构
Hadoop HDFS
简介
Hadoop Distributed File System，分布式文件系统
架构
- Block 数据块
1：基本存储单位，一般大小为64M（配置大的块主要是因为：
1）减少搜寻时间，一般硬盘传输速率比寻道时间要快，大的块可以减少寻道时间；
2）减少管理数据块的开销，每个块都需要NameNode上有对应的记录；
3）对数据块进行读写，减少建立网络的连接成本。
2：一个大文件会被拆分为一个个的块，然后存储于不同的机器，如果文件的大小 小于 Block的大小，那么实际的占用空间就为文件的大小。
3：基本的读写单位，类似于磁盘的页，每次都是读写一个块
4：每个块都会被复制到对台机器，默认三份
HDFS2.x以后的block默认128M
-
NameNode
1：存储文件的metadata，运行时所有数据都保存在内存，整个HDFS可存储的文件大小受限于NameNode的内存大小
2：一个Block在NameNode中对应一条记录（一般一个block占用150字节），如果是大量的小文件，会消耗大量内存。同时map task的数量是由splits来决定的，所以用MapReduce处理大量的小文件时，就会产生过多的map task，线程管理开销将会增加作业时间。处理大量小文件的速度远远小于处理同等大小的大文件的速度。因此Hadoop建议存储大文件
3：数据会定时保存到本地磁盘，但不保存block的位置信息，而是由DataNode注册时上报和运行时维护（NameNode中与DataNode相关的信息并不保存到NameNode的文件系统中，而是NameNode每次重启后，动态重建）
4：NameNode失效则整个HDFS都失效了，所以要保证NameNode的可用性 -
Secondary NameNode
1：定时与NameNode进行同步（定期合并文件系统镜像和编辑日志，然后把合并后的传给NameNode，替换其镜像，并清空编辑日志，类似于CheckPoint机制），但NameNode失效后仍需要手工将其设置成主机 -
DataNode
1：保存具体的block数据
2：负责数据的读写和复制操作
3：DataNode启动时会向NameNode发送数据块信息，后续也会定时报告修改信息
4：DataNode之间会保持通信，复制数据块，保证数据的冗余性
Hadoop写文件
1.客户端将文件写入本地磁盘的 HDFS Client 文件中
2.当临时文件大小达到一个 block 大小时，HDFS client 通知 NameNode，申请写入文件
3.NameNode 在 HDFS 的文件系统中创建一个文件，并把该 block id 和要写入的 DataNode 的列表返回给客户端
4.客户端收到这些信息后，将临时文件写入 DataNodes
4.1 客户端将文件内容写入第一个 DataNode（一般以 4kb 为单位进行传输）
4.2 第一个 DataNode 接收后，将数据写入本地磁盘，同时也传输给第二个 DataNode
4.3 依此类推到最后一个 DataNode，数据在 DataNode 之间是通过 pipeline 的方式进行复制的
4.4 后面的 DataNode 接收完数据后，都会发送一个确认给前一个 DataNode，最终第一个 DataNode 返回确认给客户端
4.5 当客户端接收到整个 block 的确认后，会向 NameNode 发送一个最终的确认信息
4.6 如果写入某个 DataNode 失败，数据会继续写入其他的 DataNode。
然后 NameNode 会找另外一个好的 DataNode 继续复制，以保证冗余性
4.7 每个 block 都会有一个校验码，并存放到独立的文件中，以便读的时候来验证其完整性
5.文件写完后（客户端关闭），NameNode 提交文件（这时文件才可见，如果提交前，NameNode 垮掉，那文件也就丢失了。
fsync：只保证数据的信息写到 NameNode 上，但并不保证数据已经被写到DataNode 中）
- Rack aware（机架感知）
通过配置文件指定机架名和 DNS 的对应关系
假设复制参数是3，在写入文件时，会在本地的机架保存一份数据，然后在另外一个机架内保存两份数据（同机架内的传输速度快，从而提高性能）
整个 HDFS 的集群，最好是负载平衡的，这样才能尽量利用集群的优势
Hadoop 读文件
1：客户端向NameNode发送读取请求
2：NameNode返回文件的所有block和这些block所在的DataNodes（包括复制节点）
3：客户端直接从DataNode读取数据，如果该DataNode读取失败（DataNode失效或者校验码不正确），则从复制节点中读取（如果读取的数据就在本机，则直接读取，否则通过网络读取）
Hadoop 可靠性
HDFS的可靠性主要有一下几点：
冗余副本策略
可以在hdfs-site.xml 中设置复制因子指定副本数量
所有的数据块都可以复制
DataNode启动时，遍历本地文件系统，产生一份HDFS数据块和本地文件的对应关系列表（blockreport）汇报给NameNode
机架策略
HDFS的“机架感知”是通过节点之间发送一个数据包，来感应他们是否在同一个机架
一般在本机架放一个副本，再在其他机架存放一个副本，这样就可以防止机架失效时丢失数据，也可以提高带宽的利用率
心跳机制
NameNode周期性的从DataNode接受心跳信息和块报告
NameNode根据块报告验证元数据
没有按时发送心跳的DataNode会被认为是宕机，不会再给他任何I/O请求
如果DataNode失效造成副本数据数量减少，且低于预先设定的值，NameNode会检测出这些数据库，并在合适的时机重新复制这些数据库，引发重新复制的原因还包括数据副本本身损坏,磁盘错误,复制因子被增大等
安全模式
NameNode启动时会先经过一个“安全模式”
安全模式不会产生数据写
在此阶段，NameNode会收集各个DataNode发来的报告，当数据块达到最小副本数以上时，会被认为是“安全”的
在一定比例（可以设置）的数据块被认定为“安全”过后，再过若干时间，安全模式结束
当检测副本数不足的数据块时，该块会被复制，直到达到最小的副本数
校验和
在文件创立时，每个数据块都会产生数据校验和
校验和会单独作为一个隐藏文件保存在命名空间下
客户端读取数据块时，会检查校验和是否相同，发现数据块是否损坏
如果是当前读取的数据块损坏，那么就读取其他的副本
回收站
删除文件时，其实是放入回收站/trash
回收站里面的文件是可以快速恢复的
可以设置一个时间值，当回收站里面的文件超过了这个时间，就会彻底删除，并释放占用的数据块
元数据保护
映像文件和事物日志是NameNode的核心数据，可以配置为多个副本
副本会降低NameNode的处理速度，但是会增加安全性
NameNode依然是单点，如果发生故障要收工切换
快照机制
HDFS的快照（snapshot）是在某一时间点对指定文件系统进行拷贝，快照采用只读模式，可以对重要的数据进行恢复，防止用户错误性的操作
快照分两种：
（1）建立文件系统的索引，每次更新文件不会真正的改变文件，而是新开辟一个空间用来保存更改的文件，
（2）拷贝所有的文件系统。HDFS属于前者。
HDFS的快照的特征如下：
- 快照的创建是瞬间的，代价为O(1)，取决于子节点扫描文件目录的时间。
- 当仅当做快照的文件目录下有文件更新时才会占用小部分内存，占用内存的大小为O(M)，其中M为更改文件或者目录的数量；
- 新建快照的时候，Datanode中的block不会被复制，快照中只是记录了文件块的列表和大小信息
- 快照不会影响正常的hdfs的操作。对做快照之后的数据进行的更改将会按照时间顺序逆序的记录下来，用户访问的还是当前最新的数据，快照里的内容为快照创建的时间点时文件的内容减去当前文件的内容。
- 每个快照最高限额为65536个文件或者文件夹，在快照的子文件夹中不允许在创建新的快照。
Hadoop 命令工具
fask：检查文件的完整性
start-balancer.sh: 重新平衡HDFS
hdfs dfs -copyFromLocal 从本地磁盘复制文件到HDFS
Hadoop YARN
Hadoop ResourceManager
负责全局的资源管理和任务调度，把整个集群当成计算资源池，只关注分配，不管应用，且不负责容错
资源管理
1：以前资源是每个节点分成一个个的Map slot和Reduce slot，现在是一个个Container，每个Container可以根据需要运行ApplicationMaster、Map、Reduce或者任意的程序
2：以前的资源分配是静态的，目前是动态的，资源利用率更高
3：Container是资源申请的单位，一个资源申请格式：<resource-name, priority, resource-requirement, number-of-containers>, resource-name：主机名、机架名或*（代表任意机器）, resource-requirement：目前只支持CPU和内存
4：用户提交作业到ResourceManager，然后在某个NodeManager上分配一个Container来运行ApplicationMaster，ApplicationMaster再根据自身程序需要向ResourceManager申请资源
5：YARN有一套Container的生命周期管理机制，而ApplicationMaster和其Container之间的管理是应用程序自己定义的
任务调度
1：只关注任务的调度，根据需求合理分配资源
2：Scheluer可以根据申请的需要，在特定的机器上申请特定的资源（ApplicationMaster负责申请资源时的数据本地化的考虑，ResourceManager将尽量满足其申请需求，在指定的机器上分配Container，从而减少数据移动）
内部结构
-
Client Service: 应用提交、终止、输出信息（应用、队列、集群等的状态信息）
-
Adaminstration Service: 队列、节点、Client权限管理
-
ApplicationMasterService: 注册、终止ApplicationMaster, 获取ApplicationMaster的资源申请或取消的请求，并将其异步地传给Scheduler, 单线程处理
-
ApplicationMaster Liveliness Monitor: 接收ApplicationMaster的心跳消息，如果某个ApplicationMaster在一定时间内没有发送心跳，则被任务失效，其资源将会被回收，然后ResourceManager会重新分配一个ApplicationMaster运行该应用（默认尝试2次）
-
Resource Tracker Service: 注册节点, 接收各注册节点的心跳消息
-
NodeManagers Liveliness Monitor: 监控每个节点的心跳消息，如果长时间没有收到心跳消息，则认为该节点无效, 同时所有在该节点上的Container都标记成无效，也不会调度任务到该节点运行
-
ApplicationManager: 管理应用程序，记录和管理已完成的应用
-
ApplicationMaster Launcher: 一个应用提交后，负责与NodeManager交互，分配Container并加载ApplicationMaster，也负责终止或销毁
-
YarnScheduler: 资源调度分配， 有FIFO(with Priority)，Fair，Capacity方式
-
ContainerAllocationExpirer: 管理已分配但没有启用的Container，超过一定时间则将其回收
NodeManager
Node节点下的Container管理
1：启动时向ResourceManager注册并定时发送心跳信息，等待ResourceManager的指令
2：监控Container的运行，维护Container的生命周期，监控Container的资源使用情况
3：启动或停止Container，管理任务运行时的依赖包（根据ApplicationMaster的需要，启动Container之前将需要的程序及其依赖包、配置文件等拷贝到本地）
内部结构
-
NodeStatusUpdater: 启动向ResourceManager注册，报告该节点的可用资源情况，通信的端口和后续状态的维护
-
ContainerManager: 接收RPC请求（启动、停止），资源本地化（下载应用需要的资源到本地，根据需要共享这些资源）
PUBLIC: /filecachePRIVATE: /usercache//filecache
APPLICATION: /usercache//appcache//（在程序完成后会被删除）
-
ContainersLauncher: 加载或终止Container
-
ContainerMonitor: 监控Container的运行和资源使用情况
-
ContainerExecutor: 和底层操作系统交互，加载要运行的程序
Hadoop ApplicationMaster
单个作业的资源管理和任务监控
具体功能描述：
- 计算应用的资源需求，资源可以是静态或动态计算的，静态的一般是Client申请时就指定了，动态则需要ApplicationMaster根据应用的运行状态来决定
- 根据数据来申请对应位置的资源（Data Locality）
- 向ResourceManager申请资源，与NodeManager交互进行程序的运行和监控，监控申请的资源的使用情况，监控作业进度
- 跟踪任务状态和进度，定时向ResourceManager发送心跳消息，报告资源的使用情况和应用的进度信息
- 负责本作业内的任务的容错
ApplicationMaster可以是用任何语言编写的程序，它和ResourceManager和NodeManager之间是通过ProtocolBuf交互，以前是一个全局的JobTracker负责的，现在每个作业都一个，可伸缩性更强，至少不会因为作业太多，造成JobTracker瓶颈。同时将作业的逻辑放到一个独立的ApplicationMaster中，使得灵活性更加高，每个作业都可以有自己的处理方式，不用绑定到MapReduce的处理模式上
如何计算资源需求
一般的MapReduce是根据block数量来定Map和Reduce的计算数量，然后一般的Map或Reduce就占用一个Container
如何发现数据的本地化
数据本地化是通过HDFS的block分片信息获取的
Hadoop Container
- 基本的资源单位（CPU、内存等）
- Container可以加载任意程序，而且不限于Java
- 一个Node可以包含多个Container，也可以是一个大的Container
- ApplicationMaster可以根据需要，动态申请和释放Container
Hadoop Failover
失败类型
1：程序问题
2：进程问题
3：硬件问题
失败处理
任务失败
- 运行时异常或者JVM退出都会报告给ApplicationMaster
- 通过心跳来检查挂住的任务(timeout)，会检查多次（可配置）才判断该任务是否失效
- 一个作业的任务失败率超过配置，则认为该作业失败
- 失败的任务或作业都会有ApplicationMaster重新运行
ApplicationMaster失败
- ApplicationMaster定时发送心跳信号到ResourceManager，通常一旦ApplicationMaster失败，则认为失败，但也可以通过配置多次后才失败
- 一旦ApplicationMaster失败，ResourceManager会启动一个新的ApplicationMaster
- 新的ApplicationMaster负责恢复之前错误的ApplicationMaster的状态(yarn.app.mapreduce.am.job.recovery.enable=true)，这一步是通过将应用运行状态保存到共享的存储上来实现的，ResourceManager不会负责任务状态的保存和恢复
- Client也会定时向ApplicationMaster查询进度和状态，一旦发现其失败，则向ResouceManager询问新的ApplicationMaster
NodeManager失败
- NodeManager定时发送心跳到ResourceManager，如果超过一段时间没有收到心跳消息，ResourceManager就会将其移除
- 任何运行在该NodeManager上的任务和ApplicationMaster都会在其他NodeManager上进行恢复
- 如果某个NodeManager失败的次数太多，ApplicationMaster会将其加入黑名单（ResourceManager没有），任务调度时不在其上运行任务
ResourceManager失败
- 通过checkpoint机制，定时将其状态保存到磁盘，然后失败的时候，重新运行
- 通过zookeeper同步状态和实现透明的HA
可以看出，一般的错误处理都是由当前模块的父模块进行监控（心跳）和恢复。而最顶端的模块则通过定时保存、同步状态和zookeeper来?实现HA
Hadoop MapReduce
简介
一种分布式的计算方式指定一个Map函数，用来把一组键值对映射成一组新的键值对，指定并发的Reduce（规约）函数，用来保存映射的键值对中的每一个共享相同键组
Pattern
map: (K1, V1) → list(K2, V2) combine: (K2, list(V2)) → list(K2, V2) reduce: (K2, list(V2)) → list(K3, V3)
Map输出格式和Reduce输入格式一定是相同的
基本流程
MapReduce主要是读取文件数据，然后进行map处理，再然后进行Reduce处理，最后把处理结果写到文件中
详细流程
多节点下的流程
主要过程
Map Side
Record reader
记录阅读器会翻译由输入格式生成的记录，记录阅读器用于将数据解析给记录，并不分析记录自身。记录读取器的目的是将数据解析成记录，但不分析记录本身。它将数据以键值对的形式传输给mapper。通常键是位置信息，值是构成记录的数据存储块.自定义记录不在本文讨论范围之内.
Map
在映射器中用户提供的代码称为中间对。对于键值的具体定义是慎重的，因为定义对于分布式任务的完成具有重要意义.键决定了数据分类的依据，而值决定了处理器中的分析信息.本书的设计模式将会展示大量细节来解释特定键值如何选择.
Shuffle and Sort
ruduce任务以随机和排序步骤开始。此步骤写入输出文件并下载到本地计算机。这些数据采用键进行排序以把等价密钥组合到一起。
Reduce
reduce采用分组数据作为输入。该功能传递键和此键相关值的迭代器。可以采用多种方式来汇总、过滤或者合并数据。当reduce功能完成，就会发送0个或多个键值对。
输出格式
输出格式会转换最终的键值对并写入文件。默认情况下键和值以tab分割，各记录以换行符分割。因此可以自定义更多输出格式，最终数据会写入HDFS。类似记录读取，自定义输出格式不在本书范围。
Hadoop 读取数据
MapReduce-读取数据
通过InputFormat决定读取的数据的类型，然后拆分成一个个InputSplit，每个InputSplit对应一个Map处理，RecordReader读取InputSplit的内容给Map
InputFormat
决定读取文件的格式，可以是文件或者数据库
功能
-
验证作业输入的正确性，如格式等
-
将输入文件切割成逻辑分片(InputSplit)，一个InputSplit将会被分配给一个独立的Map任务
-
提供RecordReader实现，读取InputSplit中的"K-V对"供Mapper使用
方法
List getSplits(): 获取由输入文件计算出输入分片(InputSplit)，解决数据或文件分割成片问题
RecordReader createRecordReader(): 创建RecordReader，从InputSplit中读取数据，解决读取分片中数据问题
类结构
TextInputFormat: 输入文件中的每一行就是一个记录，Key是这一行的byte offset，而value是这一行的内容
KeyValueTextInputFormat: 输入文件中每一行就是一个记录，第一个分隔符字符切分每行。在分隔符字符之前的内容为Key，在之后的为Value。分隔符变量通过key.value.separator.in.input.line变量设置，默认为(\t)字符。
NLineInputFormat: 与TextInputFormat一样，但每个数据块必须保证有且只有Ｎ行，mapred.line.input.format.linespermap属性，默认为１
SequenceFileInputFormat: 一个用来读取字符流数据的InputFormat，<key,value>为用户自定义的。字符流数据是Hadoop自定义的压缩的二进制数据格式。它用来优化从一个MapReduce任务的输出到另一个MapReduce任务的输入之间的数据传输过程。</key,value>
InputSplit
代表一个个逻辑分片，并没有真正存储数据，只是提供了一个如何将数据分片的方法
Split内有Location信息，利于数据局部化
一个InputSplit给一个单独的Map处理
public abstract class InputSplit {
/**
* 获取Split的大小，支持根据size对InputSplit排序.
*/
public abstract long getLength() throws IOException, InterruptedException;
/**
* 获取存储该分片的数据所在的节点位置.
*/
public abstract String[] getLocations() throws IOException, InterruptedException;
}
RecordReader
将InputSplit拆分为一个个<key,value>交给Map处理，也是实际的文件读取分隔对象</key,value>
问题
大量小文件如何处理
CombineFileInputFormat可以将若干个Split打包成一个，目的是避免过多的Map任务（因为Split的数目决定了Map的数目，大量的Mapper Task创建销毁开销将是巨大的）
怎么计算split的
通常一个split就是一个block（FileInputFormat仅仅拆分比block大的文件），这样做的好处是使得Map可以在存储有当前数据的节点上运行本地的任务，而不需要通过网络进行跨节点的任务调度
通过mapred.min.split.size， mapred.max.split.size, block.size来控制拆分的大小
如果mapred.min.split.size大于block size，则会将两个block合成到一个split，这样有部分block数据需要通过网络读取
如果mapred.max.split.size小于block size，则会将一个block拆成多个split，增加了Map任务数（Map对split进行计算并且上报结果，关闭当前计算打开新的split均需要耗费资源）
先获取文件在HDFS上的路径和Block信息，然后根据splitSize对文件进行切分（ splitSize = computeSplitSize(blockSize, minSize, maxSize) ），默认splitSize 就等于blockSize的默认值（64m）
public List<InputSplit> getSplits(JobContext job) throws IOException {
// 首先计算分片的最大和最小值。这两个值将会用来计算分片的大小
long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
long maxSize = getMaxSplitSize(job);
// generate splits
List<InputSplit> splits = new ArrayList<InputSplit>();
List<FileStatus> files = listStatus(job);
for (FileStatus file: files) {
Path path = file.getPath();
long length = file.getLen();
if (length != 0) {
FileSystem fs = path.getFileSystem(job.getConfiguration());
// 获取该文件所有的block信息列表[hostname, offset, length]
BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);
// 判断文件是否可分割，通常是可分割的，但如果文件是压缩的，将不可分割
if (isSplitable(job, path)) {
long blockSize = file.getBlockSize();
// 计算分片大小
// 即 Math.max(minSize, Math.min(maxSize, blockSize));
long splitSize = computeSplitSize(blockSize, minSize, maxSize);
long bytesRemaining = length;
// 循环分片。
// 当剩余数据与分片大小比值大于Split_Slop时，继续分片， 小于等于时，停止分片
while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
splits.add(makeSplit(path, length-bytesRemaining, splitSize, blkLocations[blkIndex].getHosts()));
bytesRemaining -= splitSize;
}
// 处理余下的数据
if (bytesRemaining != 0) {
splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining, blkLocations[blkLocations.length-1].getHosts()));
}
} else {
// 不可split，整块返回
splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts()));
}
} else {
// 对于长度为0的文件，创建空Hosts列表，返回
splits.add(makeSplit(path, 0, length, new String[0]));
}
}
// 设置输入文件数量
job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
LOG.debug("Total # of splits: " + splits.size());
return splits;
}