Yige

Hadoop 系列 - HDFS#

一、HDFS 的启动过程#

1. 加载文件的元信息
2. 加载日志文件
3. 设置检查点
4. 进入安全模式。作用是：检查数据块的副本率，冗余度是否达到要求

二、HDFS 的运行机制#

用户文件会被切块后存储在多台 DataNode 服务器当中，并且每个文件在整个集群中存放多个副本，可以提高数据的安全性

三、基本架构#

• NameNode: 整个文件系统的管理节点，负责记录文件如何被切分成 Block 数据块，同时记录这些数据块的存储信息

  1. Fsimage: 元数据存储在硬盘上的镜像文件
  2. edits: 系统操作的日志记录文件
  3. Fstime: 保存最近一次 checkpoint 的时间
  4. seen_txid: 最后一个 edites 的数字
  5. version

• Secondary NameNode: 辅助后台程序 (不是 NameNode 的容灾节点)，与 NameNode 进行通信，以便定期保存 HDFS 元数据的快照

• DataNode: 数据节点，负责把 HDFS 数据块读写到本地的文件系统

HDFS 不适合存储小文件的原因，每个文件都会产生元信息，当小文件多了之后元信息也就多了，对 namenode 会造成压力

联邦 HDFS#

每个 NameNode 维护一个命名空间，不同 NameNode 之间的命名空间相互独立，每个 DataNode 需要注册到每个 namenode 上

• 多个 NN 共用一个集群 DN 的存储资源，每个 NN 都可以单独对外提供服务。

• 每个 NN 都会定义一个存储池，有单独的 id，每个 DN 都为所有存储池提供存储。

• DN 会按照存储池 id 向其对应的 NN 汇报块信息，同时，DN 会向所有 NN 汇报本地存储可用资源情况。

• 如果需要在客户端方便的访问若干个 NN 上的资源，可以使用客户端挂载表，把不同的目录映射到不同的 NN，但 NN 上必须存在相应的目录

四、NameNode 工作机制#

HDFS 读写 -> 滚动记录日志 -> SN 向 NN 询问是否需要 checkPoint -> 时间到了 (60 分钟) 或者 edits 数据满了触发检查点 -> SN 请求执行 checkPoint -> NN 拷贝 edits 文件和 fsimag 文件到 SN -> SN 端合并 editslog 到 fsimage -> SN 将计算合并完的 fsimage 同步到 NN

五、DataNode 工作机制#

1. DataNode 启动向 NN 注册，然后定期汇报 block 数据块的信息
2. NN 和 DataNode 之间通过心跳检测机制，心跳 1 次 / 3s，如果超过 10 分钟未收到心跳则认为节点不可用

六、HDFS 读数据流程#

Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(conf);
Path file = new Path("demo.txt");
FSDataInputStream inStream = fs.open(file);
String data = inStream.readUTF();
System.out.println(data);
inStream.close();

1. 客户端初始化FileSystem对象，调用open()方法获取一个DistributedFileSystem 对象
2. DistributedFileSystem 通过 RPC 向 NN 请求获取第一批 block locations
3. 前两部会生成一个 FSDataInputStream, 该对象会被封装成 DFSInputStream 对象
4. 客户端调用read()方法，DFSInputStream 会找出离客户端最近的 Datanode 并连接开始读取此文件的第一个数据块，数据从 Datanode 传输向客户端
5. 当第一个数据块读取完毕，DFSInputStream 关闭连接，然后连接下一次数据块的 DataNode 节点继续数据传输
6. 如果在读数据的时候 DFSInputStream 和 Datanode 数据节点的通讯发生异常，就会尝试连接下一个包含次数据块的 DataNode 节点，并且会记录哪个 datanode 发生错误，剩余的 blocks 读的时候就会直接跳过该 datanode
7. 客户端读取数据完毕后调用 close () 方法关闭连接然后将数据写入到本地文件系统

七、HDFS 写数据流程#

Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(conf);
Path file = new Path("demo.txt");
FSDataOutputStream outStream = fs.create(file);
out.write("Welcome to HDFS Java API !!!".getBytes("UTF-8"));
outStream.close();

1. 客户端调用的create方法，创建一个文件输出流DFSDataOutputStream对象，向 NN 请求上传文件
2. DistributedFileSystem 通过 RPC 调用 NN 去创建一个没有 blocks 关联的新文件条目（Entry）, 创建前 NN 会校验文件是否存在或者是否有权限创建，成功直接先将操作写入 EditLog (WAL，write ahead log)，并返回输出流对象，否则抛出 IO 异常
3. 客户端会首先将文件进行拆分，比如一个 Block 块 128M，文件有 300M 会被切分为 3 个块，两个 128M，一个 44M，然后向 NN 请求 Block 该传输到哪些 DataNode 服务器上
4. NN 返回可写的 DN 节点信息，客户端和 NameNode 分配的多个 DataNode 构成 pipeline 管道建立起连接，client 端向输出流对象中写数据
5. 通过 FSDataOutputStream 对象向 DN 写数据，这些数据会被拆分成一个个小的 packet，然后排成队列 data quene。客户端每向第一个 DataNode 写入一个 packet，这个 packet 便会直接在 pipeline 里传给第二个、第三个…DataNode，并不是写好一个块或一整个文件后才向后分发
6. 每个 DN 写完一个 block 后向客户端响应ack确认信息, 注意并不是每写完一个 packet 后就返回确认信息
7. 客户端完成写数据后调用close方法关闭流

补充:

• 客户端执行 write 操作后，写完的 block 才是可见的，正在写的 block 对客户端是不可见的，只有调用 sync方法，客户端才确保该文件的写操作已经全部完成，当客户端调用 close 方法时，会默认调用 sync 方法。是否需要手动调用取决你根据程序需要在数据健壮性和吞吐率之间的权衡

写的过程中错误处理:#

写的过程某个 DN 副本节点错误:#

1. 首先 pipeline 会被关闭掉
2. 将已经发送到管道但还没有收到确认的 packet 写回到 data quene, 避免数据丢失
3. 然后当前正常工作的 DN 数据节点将会被赋予一个新的版本号（利用 namenode 中租约的信息可以获得最新的时间戳版本），这样故障节点恢复后由于版本信息不对，故障 DataNode 恢复后会被删除
4. 删除故障节点，选择正常的 DN 数据节点重新建立管道，然后开始重新写入数据
5. 发现副本不足，NN 会去其他的 DN 节点上创建一个新的副本

写的过程客户端崩溃解决:#

当数据写入过程中客户端异常退出时，同一 block 数据块的不同副本可能存在不一致的状态。 选择某一副本作为主数据节点，协调其他数据节点，NN 会通过租约机制 (lease) 找到所有 DN 副本节点拥有这个数据块信息的最小 block 长度，然后将该数据块恢复到他们中的最小长度

详细参考: HDFS 恢复过程 1

八、HDFS 副本机制#

第一副本：如果上传节点是 DN，则上传该节点；如果上传节点是 NN，则随机选择 DN
第二副本：放置在不同机架的 DN 上
第三副本：放置在与第二副本相同机架的不同 DN 上

九、HDFS 安全模式#

安全模式是 HDFS 的一种工作状态，处于安全模式的状态下，只向客户端提供文件的只读视图，不接受对命名空间的修改

• NN 启动时首先会将 fsimage 加载到内存中，然后执行 edits 日志记录中的操作，当在内存中成功建立文件系统元数据的映射后，新建一个 fsimage 和一个 edits 空的编辑日志文件，这个时候 NN 运行在安全模式
• 在此阶段 NN 通过 DN 收集信息，对每个文件的数据块进行统计，当确认满足最小副本条件时，即一定比例的数据块都达到最小副本数，就会退出安全模式。当不满足的时候安排 DN 对对副本数不足的数据块进行复制，直至达到最小副本数
• 在启动一个刚刚格式化的 HDFS 时不会进入安全模式，因为没有数据块

退出安全模式：hdfs namenode -safemode leave

十、HA 高可用机制#

参考: Hadoop NameNode 高可用 (High Availability) 实现解析

基本架构实现#

HDFS 的 HA 高可用通过 zk 保证，基本架构:

• Active NameNode 和 Standby NameNode: 主备 NameNode 节点，只有处于 Active 的主 NameNode 节点对外提供服务

• 共享存储系统: 保存了 NN 运行过程中产生的元数据，主备 NN 通过共享存储系统实现元数据同步，在进行主备切换的时候新的 NN 只有确认元数据完全同步后才能对外提供服务

• 主备切换控制器 ZKFailoverController ： ZKFC 作为独立的进程运行，能及时监测到 NN 的健康状态，当主 NN 出现故障时借助 zk 集群实现主备自动选举切换

• DataNode： DataNode 需要同时向主备 NN 上传数据块信息保证 HDFS 的数据块和 DataNode 之间的映射关系同步

• Zookeeper 集群：为主备切换控制器提供主备选举支持

主备切换实现#

参考: Hadoop NameNode 高可用 (High Availability) 实现解析

NameNode 主备切换主要由ZKFailoverController、HealthMonitor 和 ActiveStandbyElector 这 3 个组件来协同实现：

• 主备切换控制器 ZKFailoverController 启动的时候会创建 HealthMonitor 和 ActiveStandbyElector 这两个主要的内部组件，ZKFailoverController 在创建 HealthMonitor 和 ActiveStandbyElector 的同时，也会向 HealthMonitor 和 ActiveStandbyElector 注册相应的回调方法。

• HealthMonitor 主要负责检测 NameNode 的健康状态，如果检测到 NameNode 的状态发生变化，会回调 ZKFailoverController 的相应方法进行自动的主备选举。

• ActiveStandbyElector 主要负责完成自动的主备选举，内部封装了 Zookeeper 的处理逻辑，一旦 Zookeeper 主备选举完成，会回调 ZKFailoverController 的相应方法来进行 NameNode 的主备状态切换

如图，流程分析:

1. HealthMonitor 初始化完成后会启动内部线程定时调用对应 NameNode 的 HAServiceProtocol RPC 接口的方法，对 NameNode 的健康状态进行检测
2. 当检测到 NN 状态发生变化时，回调 ZKFailoverController 的相应方法进行处理
3. ZKFailoverController 监测到需要进行主备切换时使用 ActiveStandbyElector 进行处理
4. ActiveStandbyElector 与 ZK 进行交互完成自动选举，然后回调 ZKFailoverController 的相应方法通知当前 NN
5. ZKFailoverController 调用对应 NameNode 的 HAServiceProtocol RPC 接口的方法将 NameNode 转换为 Active 状态或 Standby 状态