HDFS介绍

HDFS介绍

Trafalgar

HDFS是Hadoop使用的分布式文件系统，能存储和处理大规模数据。HDFS的设计目标是在标准硬件上运行，从而提供高容错性，并且能够处理已存储的大量数据。

使用场景

首先需要明确的是，所有的存储都是为计算服务的。

计算任务根据其实时性可以分为两种，在线计算和离线计算。

在线计算的实时性要求比较强，所以对存储的时延要求高，往往会存储在MySQL之类的TP数据库中。

离线计算对实时性要求比较低，所以对存储的时延要求低，一般只要求能正常读写就可以了，会更加关注存储的成本。这一类数据会选择存储在HDFS上。

HDFS通常用于处理离线数据的存储和分析，例如Web日志数据或者机器学习训练数据。另外，由于HDFS天生就是Hadoop生态系统的存储底座，它也可以作为Hadoop生态系统中其他工具的基础，例如MapReduce和Spark，所以可以支持的上层计算引擎非常丰富。

总体架构

HDFS在编写的时候主要是参考GFS，所以架构比较经典。主要分为Client，Namenode和Datanode三个组件。Namenode的HA使用主从备份方案，Datanode使用三副本的链式复制。

HDFS集群由一个NameNode和多个DataNode组成。NameNode负责管理文件系统的命名空间（Namespace），存储文件的元数据信息（如文件名、文件路径、文件长度、文件块列表等），以及每个文件块所在的DataNode节点。DataNode负责存储实际的数据块（Block），并定期向NameNode汇报自己所存储的数据块列表。

HDFS分离元数据和实际数据的设计，使得NameNode可以专注于管理文件系统的命名空间，Datanode只负责数据存储。理论上可以通过横向拓展实现无限存储。

客户端

DFSClient是所有客户端底层调用的对象。

读数据

1. 底层调用DFSClient.open()打开文件，并构造文件输入流。构建文件流时，需要调用openInfo找Namenode拿到该文件对应的Block信息。再从Datanode拿到最后一个Block的长度信息。
2. 读分为以下三种，依次速度增加。在进行读取时，会优先使用最快的

1. 网络读会根据当前offset从Namenode找到对应Block存储的Datanode，从中选出一个最近的，然后构建BlockReader对象。如果使用BlockReader读取校验失败，会汇报给Namenode。
2. BlockReader会根据参数生成网络读或者短路读的Reader。RemoteBlockReader2用于网络读，会使用PacketReceiver.receiveNextPacket()来读取Packet，就是根据头部读出数据，并进行校验
3. BlockReaderLocal用于进行本地短路读，会使用Domain Socket来进行通信，还使用了双buffer的策略，保证每次都能完整地进行校验。
4. tryReadZeroCopy()是用于进行零拷贝读的方法，底层是调用linux的系统调用。如果读取失败，会fallback到上面两种读取。
5. BlockReaderLocal用于进行本地短路读，会使用Domain Socket来进行通信，还使用了双buffer的策略，保证每次都能完整地进行校验。
6. tryReadZeroCopy()是用于进行零拷贝读的方法，底层是调用linux的系统调用。如果读取失败，会fallback到上面两种读取。

3.文件短路读操作

1. 同一个物理机上的Client和Datanode可以通过Domain Socket，使用共享内存进行数据传输。两者通过DataTransferProtocol约定协商Slot对象，Slot在共享内存中分配，就是一个Slot数组，是管理内存的基本单元。结构如下。

1. 流程是先通过requestShortCircuitShm()申请供销内存，创建完毕后，客户端创建Slot，并调用requestShortCircuitFds()向Datanode发起打开文件的请求，并通过Domain Socket来传输文件描述符。DFSClient 的 BlockReaderFactory对象成功地从 domainSocket 接收了数据块文件和校验和文件的文件描述符之后，就可以初始化数据块文件和校验和文件的输入流并构造ShortCircuitReplica对象了，然后通过文件流来读取文件。Datanode会根据客户端的Slot创建自己的Slot。
2. Slot的Anchor是一个整型，表示该块内存无需验证，同时记录了该内存的引用计数。
3. 在客户端，会为每个Datanode维护一个DfsClientShmManager，这个Manager里面维护了两个DfsClientShm队列，一个是内存已满的，一个是内存未满的。ShortCircuitReplica类封装了一个短路读数据块副本的所有信息。ShortCircuitCache则维护了负责对 DFSClient 的所有ShortCircuitReplica 对象进行缓存以及生命周期管理等操作。其周期转化如下图所示。
4. 在Datanode端，维护了一个ShortCircuitRegistry来管理所有的共享内存，共享内存的抽象是RegisteredShm。

3.文件写操作

1. 对于create写，需要从Namenode拿到文件状态，然后计算这次写的packet大小，一般是65532B。如果有错误，则将ackQueue中所有的Packet都重新发送。所有写入完毕后，发送空包，表示结束。其发送流程如下图所示。

1. 写入时，先使用Packet对象缓存数据。Packet写满时，将其放入dataQueue，等待DataStreamer取出发送。有可能最后一个chunk写不满，需要对其进行paddingf。写入流程如下图所示。

1. 实际执行写入的是DataStreamer，会调用 nextBlockOutputStream()方法向 Namenode 申请 分配新的数据块，然后根据数据块信息构建输出流，底层是使用Sender对象。如果是最后一个包，需要等前面所有Packet都发送才能发送，并且需要等待其完成。
2. Append写也是类似，只是需要打开最后一个Block的流。如果最后一个Block已满，则按照正常逻辑打开。

Namenode

1.文件系统目录树

INode相关

1. INode有两个关键子类，INodeDirectory和INodeFile。
2. INode的成员变量包含了POSIX标准的七个属性
3. INodeDirectory和INodeFile还实现了Permission属性和Feature属性，Feature属性中包含了磁盘配额、正在构建（UnderConstrution）、快照（Snapshot）等HDFS额外提供的功能。
4. 为了支持在快照下剪切功能，引入了WithName，WithCount和DstReference三个类。快照中的原文件将使用WithName代替，指向WithCount对象。剪切的目标位置将指向DstReference对象，DstReference也将指向WithCount对象。WithCount对象将指向真正的文件。

Snapshot

1. 快照支持对某个时刻的文件系统进行只读操作，该功能通过在Feature中添加DirectoryWithSnapshotFeature或者DirectorySnapshottableFeature来实现。
2. 在DirectoryWithSnapshotFeature中，对每个快照将维护两个list，CeateList和DeleteList，分别记录在该快照以后，对快照内的文件进行删改。
3. 添加文件时，需要在CreateList中增加该文件，不会改动原目录的结构。
4. 使用快照读时，会从当前快照向后查找，检查dlist中是否有相关文件，如果有直接返回，如果一直没有就去原目录下读。
5. 读取快照文件夹下的文件时，需要根据当前状态，clist和dlist进行逆推。

FSEditLog

1. 分为五个状态，UNINITIALIZED最初试的状态，BETWEEN_LOG_SEGMENTS已经打开但还未写入数据，IN_SEGMENT可以写入数据的状态，CLOSED关闭，OPEN_FOR_READING备份节点初始化的状态。
2. 为了处理存储在不同软件上的情况，进行了抽象，使用JournalSet进行管理。
3. 真正写入数据时，使用双Buffer进行优化，主备Buffer防止Block。其中，向buffer写数据会进行同步。由于正在写的buffer会被标记为isSyncRunning=true，真正进行Flush的时候无需同步。

FSImage

1. 很简单的东西，用protobuf进行序列化。
2. 新记录存到editlog，然后定期和fsimage进行合并。

2.数据块管理

Block

1. BlockInfo中的BlockCollection指向其属于的文件，triplets是一个隐性的双向链表，3i是自己，3i+1是同一个Datanode存储的上一个BlockInfo对象，3*i+2是后一个。而如果保存了3个副本，那么第0,3,6则是三个副本所在的Datanode位置。
2. Namenode只存储Block的信息，包括blockid，大小和时间戳。具体的Block和BlockInfo对应的关系会根据Datanode的上报进行动态更新。

BlockManager

1. 具体下发复制指令时，会从needReplications队列中选取一定数量的blocks一次性处理。这里一共设涉及三个问题，怎么选择源节点，目标节点和什么时候执行任务。源节点的选择还是随机的，但是会优先选择不在写集合里面的节点。目标节点的选择遵循不同机架，不同Datanode的原则。执行该任务是在Datanode上报信息时，在Response中携带该指令。
2. 增加/删除Block是通过INode的相关方法实现的，增加/删除Replica是通过各种的操作实现的
3. Corruption是通过比较Datanode上报信息和Namenode信息来发现的。
4. 所有副本数量的变动都会导致需要复制的副本数量变化。需要变多的，由needReplica记录，并每隔一段时间转发到pendingReplica中，在Datanode汇报时进行下发。需要变少的，放到excessReplicateMap中，每隔一段时间转发到invalidateBlocks中进行下发。
5. 数据块的汇报由Datenode发起，第一次是全量汇报，后面只汇报增量。全量处理时，会在上报的链表头放一个虚Block作为分隔符，每处理一个就放到分隔符前面。全部处理完后，在分隔符后的，就是需要删除的。

3.Datanode管理

1. 数据块和数据节点的映射，即triplet，使用Block→DataNodeStorageInfo的映射
2. Decommission的时候需要确保该Datanode上所有数据的副本数都达标才能标记为已Decommission
3. Datanode注册的时候，需要区分从未注册过，使用不同StorageId重复注册，使用相同StorageId注册的三种情况。
4. 收到Datanode心跳时，会根据其存储信息更新全局信息，检查其损坏Block和Storage，并且下发已经准备的命令。

4.租约管理

1. 租约按照文件赋予客户端，关键是LeaseManagerd的leases，sortedLeases和sortedLeasesByPath三个对象
2. 只有UnderConstruction的数据块才会有lease，已经构建完毕的都是只读的。
3. 租约恢复的过程，选取一个最近一次进行汇报的数据节点作为主恢复节点，然后向这个数据节点发送租约恢复指令，主恢复数据节点接收到指令后，会调用Datanode.recoverBlock()方法开始租约恢复，这个方法首先会向数据流管道中参与租约恢复的数据节点收集副本信息，然后从该数据块的所有副本中选取一个最好的状态作为所有副本恢复的目标状态（多个副本中选择最小长度作为最终更新一致的标准）。确后主恢复节点会同步所有Datanode上该数据块副本至目标状态。同步结束后，这些数据节点上的副本长度和时间戳将一致。最后，主恢复节点会向NameNode报告这次租约恢复的结果。NameNode 更新文件 block 元数据信息，收回该文件租约，并关闭文件。

Datanode

1.Datanode架构

hdfs2.6引入了Federation架构，为了支持拓展，Namenode可以有多个，每个管理不同的Namespace。在BlockStorage这一层引入了BlockPool。每个BlockPool对应一个Namenode管理的Namespace。

BlockPoolManager管理BlockPool并先Namenode汇报。DataBlockScanner检查数据块校验和，DirectoryScanner扫描目录与其下属的文件元数据是否对应，并且更新。DataStorage负责管理与组织 Datanode 的磁盘存储空间，管理存储空间的生命周期，包括升级、回滚、提交等操作。通过BlockPoolSliceStorage管理BlockPool。FsDataset管理数据块的所有操作，例如创建数据块文件、维护数据块文件和校验和文件的对应关系等。用FSVolumeImpl来管理每个存储目录。

2.Datanode Storage

1. 升级时需要考虑回滚，hdfs的方案是复制一份原有数据进行备份。但是，这样需要保留额外的1/2空间，hdfs使用linux硬链接的方式，将新旧版本中相同的部分指向同一个文件。
2. 文件组织架构如下，注意，pool在current的目录下，这样可以支持单个pool进行升级。每个BlockPool的数据可以分布在不同的文件目录下。

1. 所有文件目录相关的操作都可以在StorageDirectory中进行，它是Storage的内部类。
2. Datanode启动时，会扫描配置文件中指定的目录，找到每个blockpool目录。然后分析并恢复每个目录的状态，最后调用doTransition()进行启动。
3. DataStorage负责维护上述目录的组织架构，在升级或者回滚时进行文件和文件夹的链接和重命名等操作。

3.FsDatasetImpl

1. 数据块有RBW（可以写），FINALIZED（构建完成），RUR（恢复中），RWR（等待恢复），TEMPORARY（Datanode之间再在复制的副本）等状态。
2. 启动Datanode时，需要读取所有FINALIZED和RBW数据块的元信息到内存。
3. 升级后的数据块使用硬链接减少空间使用，但如果这时对该数据块进行apppend，那么会影响原有数据。此时需要copy-on-write，将现有数据复制一份再进行append。
4. FsVolumeList管理一个Datanode下的所有存储目录，每个存储目录对应的对象是FsVolumeImpl，FsVolumeImpl管理多个BlockPool，BlockPool对应的对象是BlockPoolSlice。FsDatasetImpl会管理所有的数据块。
5. 添加副本时有轮询和选择最多剩余空间两个策略。
6. FsDatasetImpl维护一个ReplicaMap，是BlockPoolID→Map<Block，ReplicaInfo>的映射，拿到对应BlockPool的Map后就能根据Block对象得到副本信息。在添加RBW和FINALIZED的数据块时会更新ReplicaMap，后台也会定期扫描数据，更新ReplicaMap。
7. 数据块将通过以下的函数在状态间进行跳转。

1. 执行Append前需要invalid cache并unlink硬链接。FINALIZED的Block也能append，但是需要把状态改成RBW。
2. FINALIZED前需要检查是否需要Recovery，需要的话先进行Recovery，不需要就直接移动下目录，更新ReplicaMap即可。
3. 启动时也需要对数据块进行恢复，调用的是initReplicaRecovery → updateReplicaUnderRecovery，主要做各种检查，并且根据所有副本的最小长度truncate数据。。
4. HDFS的Cache是指令式的cahce，由用户手动指明需要缓存哪些数据。由FsDatasetCache维护
5. FsDatasetImpl负责向Namenode汇报数据块信息。

4.BlockPoolManager

1. BlockPoolManager是BlockPool的对外接口，负责向Namenode汇报数据并执行带回的命令。

1. BPServiceActor负责和Namenode汇报数据，心跳等信息。在启动时，会连接Namenode，判断其是否是自己第一个连接的Namenode，如果是，就向其注册自己。心跳回包中还有Namenode切换的信息，需要根据Txn id以及其声明判断是否需要进行更换汇报的Namenode。
2. blockReport负责全量数据块汇报，IncrementalBlockReportManager维护了一个Map，记录增量和删除的数据块，在IncrementalBlockReportManager.sendIBRs中进行增量汇报。
3. BPOfferService更多的是一个对外的接口，包装了一下两个BPServiceActor，然后处理一下BPServiceActor传来的关于Namenode的指令。
4. BlockPoolManager也很简单，就是管理一下BPOfferService，处理Namenode的增加，删除和refresh操作，对应增加/删除/更新BPOfferService。

5.流式接口

1. 读写数据的接口在DataTransferProtocol中，请求的pb格式
2. Datanode通过DataXceiverServer监听所有请求，并创建DataXceiver响应对应请求。Hadoop很喜欢使用建立连接和处理逻辑相隔离的设计，和Bytedrive一样，是为了提高前台的处理能力。
3. DataXceiver会先解析传入的数据流，然后创建对应的对象进行处理，readop会创建BlockSender，以下列格式发送数据。

读数据

1. BlcokSender发送数据一共分为三步，首先根据传入的数据计算校验和，检验时间戳，offset等各种东西，第二步进预读取，把数据从磁盘读到内存，第三步读取第二步中读上来的数据，调用sendPacket把数据构造成上述格式，写到输出流中。
2. 可以将transferTo设置为true来使用操作系统的零拷贝特性，减少两次内存拷贝和内核态切换。但是网卡层面不支持计算校验和，所以校验需要在用户层进行。
3. 限流在DataTransferThrottler进行，原理很简单，通过计算当前总流量和流量上限，得到剩余流量，如果现在这个请求的流量大于剩余流量，就wait一段时间，如果没有，减掉这个请求流量，直接返回。

写数据

1. HDFS写入请求遵循下面的逻辑，使用Chain Replica来复制数据，写入成功后会级联返回。
2. 通过DataXceiver.writeBlock()来执行写入逻辑，构建上下游的输入输出流，用BlockReceiver转发数据。BlockReceiver中关键的方法是receivePacket，会接收上游数据并转发给下游，如果是最后一个节点，会进行落盘并返回响应。
3. 响应在PacketResponder中进行处理，使用生产者-消费者模型，会复制下游响应给上游。如果是OOB响应，则直接转发。如果有错误，转发给上游，并停止此次的发送和接收线程。如果是Packet中的最后一个chunk的响应，则调用finalizeBlock上报给Namenode。

总结

HDFS是非常经典的分布式文件系统，由于经过多年的广泛使用，已经相当稳定。由于其在Hadoop生态的独特地位，许多公司在起步阶段都会把HDFS作为离线存储的首选方案。

虽然HDFS的一些设计在如今看来有些过时，比如Namenode的主备方案可以使用Raft。但是在工程实践上的许多细节，比如Datanode链式写入时的级联报错，不持久化Block到Datanode的信息等，仍然有许多学习的价值。

编辑于 2023-04-09 17:03・IP 属地上海

HDFS

软件介绍

产品介绍