HDFS的读写流程 —— Hadoop权威指南4_hdfs 等待所有副本写入才上传下一个block

1. 客户端读取HDFS文件的流程

1.1 具体的流程

1. 初始化FileSystem，client调用FileSystem对象的open()方法，打开一个HDFS文件。实际，FileSystem对象是一个DistributedFileSystem实例

2. DistributedFileSystem通过RPC调用NameNode，获取一批文件block的位置列表。其中，每个block的副本所在的DataNode，是按照它们与客户端的距离排序的

3. DistributedFileSystem向客户端返回一个FSDataInputStream对象（open()方法的返回结果），用于数据的读取。FSDataInputStream中包含了DFSInputStream对象，它管理着DataNode和NameNode的IO

   以上对应图中的步骤1和2

4. 客户端调用FSDataInputStream对象的read()方法，实际：存有block位置的DFSInputStream会连接最近的DataNode，对该DataNode反复执行read()，从而将数据从DataNode传到client。

   ① DFSInputStream发现DataNode故障，会记住该DataNode并从最近的另一个DataNode读取副本，以后保证不会反复从该DataNode读取block
   ② DFSInputStream会对读取到的数据做checksum，如果发现数据损坏，会从其他DataNode读取副本并向NameNode通知损坏的block信息

5. DFSInputStream完成block的读取，会关闭与该DataNode的连接。接着，创建与下一个block的最佳DataNode的连接，继续进行block的读取。

6. DFSInputStream中存储的block位置可能是文件的部分block，它会按照需要从NameNode获取下一批block的位置信息。

   以上对应步骤3、4和5

7. client读取数据完毕，则调用FSDataInputStream对象的close()方法关闭输入流

1.2 架构优势

• client从NameNode获取block的位置（多副本，排序后的DataNode地址），然后直接与DataNode连接读取数据，这样的设计允许HDFS扩展到大量的并发client。
• NameNode负责响应获取block位置的请求，而不负责响应数据请求，极大减轻了NameNode的压力。避免客户端增长后，NameNode称为系统瓶颈。

1.3 Hadoop对节点距离的描述

• Hadoop集群的节点，可能位于同一机架、同一机房不同机架、不同机房
• HDFS读写文件block时，client只有读取最近的DataNode才能提高数据读写的速度
• 因此，如何定义节点间的距离非常重要 —— 使用带宽描述两个节点之间的距离
• 针对以下场景，带宽将依次递减：
  ① 同一节点上的两个进程
  ② 同一机架上的不同节点
  ③ 同一数据中心（也就是机房）的不同机架
  ④ 不同数据中心
• 两个节点间的距离，是它们到最近共同祖先的距离之和
• 因此，图示中的节点距离就非常清楚了：
  ① 节点1上的两个进程的距离为0
  ② 节点1和节点2的距离为2
  ③ 节点1和节点3的距离为4
  ④ 节点1和节点4的距离为6
  
• 注意：
  Hadoop无法自动发现网络拓扑结构，需要进行手动配置。一般情况下，默认集群同一数据中心的同一机架上

2. HDFS数据写入流程

2.1 HDFS中block、packet、chunk

• 之前有说过，HDFS中数据存储单位是block，从Hadoop 2.7.3开始，block的默认大小从64MB变成128MB
• client向DataNode、以及DataNode的pipeline之间，数据传输的第二大单位是packet，为64 kb
• client向DataNode、以及DataNode的pipeline之间，数据校验的基本单位是chunk，默认为512 byte
• 每个chunk通过校验后，会带上4 byte的校验信息。因此，实际写入packet的chunk大小为516 byte

2.2 client的数据三层缓存

• 上传的文件，是按block进行切分的。例如，一个300MB的文件，将会被切分成128MB、128Mb、44MB三个块

写过程会涉及chunk、packet、DataQueue/AckQueue这样的三层缓存：

1. 数据流入DFSOutputStream时，会先写入一个chunk大小的缓冲区。当数据写满一个chunk时，或遇到强制的flush()操作时，会计算校验和（checksum）
2. 通过数据校验的chunk会加上校验和，一起写入packet中。当packet被chunk填满时，packet会发送到DataQueue中
3. DataStreamer负责将DataQueue中的packet发送到最佳的DataNode中。同时，由于packet此时未确认写入成功，因此会被移动到AckQueue中等待写入确认。
4. 当收到足够的确认应答（ack）后，ResponseProcessor会将packet从AckQueue中移除；否则， 会将其恢复到DataQueue的最前端，以保证没有packet丢失。因此，写入成功的packet在DataQueue和AckQueue中，应该都是不存在的。
   

2.3 HDFS的写过程

1. 初始化FileSystem对象，实际是一个DistributedFileSystem实例。

2. clinet调用DistributedFileSystem的create()方法，创建一个HDFS文件。DistributedFileSystem会通过RPC调用NameNode，进行文件创建的检查：client的权限检查、是否已存在该文件等。

3. 通过检查后，NameNode会向EditLog写入一条新建文件的记录（WAL)，然后DistributedFileSystem会向client返回一个FSDataOutputStream对象；否则，文件创建失败并向client抛出IOException异常

4. FSDataOutputStream对象中包含了DFSOutputStream对象，DFSOutputStream负责管理DataNode和NameNode的IO

   对应图中的步骤1、2

5. client调用FSDataOutputStream对象的write()方法向DataNode写入数据。实际，DFSOutputStream会将数据划分成一个个packet，先将packet存如DataQueue中。

6. 由DataStreamer负责管理DataQueue，它挑选出一组适合存储副本的DataNode，并以此来要求NameNode分配适合新的block。这组DataNode之间会形成pipeline，packet通过pipeline进行传输

   • DataStreamer将packet从DataQueue发送至第一个DataNode
   • 第一个DataNode将packet发送给第二个DataNode
   • 第二个DataNode将packet发送给第三个DataNode，从而实现packet的三副本存储

7. 同时，DataStreamer还会将未确认成功的packet移入AckQueue中，只有收到三个DataNode的ack packet后，ResponseProcessor才会将packet从AckQueue中移除

   对应步骤3、4、5

8. 如果写入期间某个DataNode故障：

   • 首先会关闭pipeline，将AckQueue中所有packet都放回到DataQueue中了，以保证故障DataNode的下游DataNode不会出现数据丢失；
   • 存储在正常DataNode中的block会被做标记，并发送给NameNode，以便故障DataNode恢复时能删除不完整的block
   • 在剩下的两个DataNode中重新建立pipeline，余下的数据会继续写入正常的DataNode。
   • NameNode在观察到副本数不够时，会在另一个DataNode上创建新的副本
   • block成功写入的副本数满足最小副本数（dfs.namenode.replication.min，默认为1)，则认为写入成功。后续通过异步的副本复制，来达到副本数要求。

   故障情况的处理

9. client完成数据写入，调用DistributedFileSystem的close()方法，关闭数据流

10. DistributedFileSystem会调用通知NameNode文件写入完成前，等待确认。实际，只需要等待block满足最小副本数，就可以确认写入成功。

    对应图中的步骤6、7

3. 其他知识

3.1 关于副本放置策略

• HDFS在进行写入时，会尽量不在一个机架放置过多的副本
• 以三副本为例：
  ① 如果client本身就是一个DataNode，则将副本1写入本地；否则，在集群中随机选择一个节点。都记为 $r1/d1$
  ② 第二个副本放置在随机选择的、不同机架的DataNode上，记为$r2/d2$
  ③ 第三个副本放置在上一副本相同的机架、随机选择的DataNode上，记为$r2/d3$

3.2 关于hflush()

• 文件系统的一致性模型：文件读/写数据的可见性
• HDFS为了性能，牺牲了一些POSIX要求：当前正在写入的block，对其他reader不可见。

HDFS提供了将缓存中的数据刷新到DataNode的方法：

1. hflush()：
   ① 在Hadoop 1.x中叫做sync()，可以保证将数据写入DataNode的内存，但不保证数据写入磁盘 —— 存在断电数据丢失的风险
   ② 保证文件中到目前为止写入的数据均到达所有DataNode的pipeline中，并且对所有新的reader可见 （其实，自己并不是很理解 声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/不正经/article/detail/456163