深入理解 Hadoop HDFS 一文读懂

野性酷女 2023-06-09 03:52 158阅读 0赞

# **1. 介绍** #

在现代的企业环境中，单机容量往往无法存储大量数据，需要跨机器存储。**统一管理分布在集群上的文件系统称为分布式文件系统。**而一旦在系统中，引入网络，就不可避免地引入了所有网络编程的复杂性，例如挑战之一是如果保证在节点不可用的时候数据不丢失。

传统的网络文件系统（NFS）虽然也称为分布式文件系统，但是其存在一些限制。由于 NFS 中，文件是存储在单机上，因此**无法提供可靠性保证**，当很多客户端同时访问 NFS Server 时，很容易**造成服务器压力，造成性能瓶颈**。另外如果要对 NFS 中的文件进行操作，需要首先同步到本地，这些修改在同步到服务端之前，其他客户端是不可见的。某种程度上，NFS 不是一种典型的分布式系统，虽然它的文件的确放在远端（单一）的服务器上面。

![20160715085646356][]

![20160715085700575][]

从 NFS 的协议栈可以看到，它事实上是一种 VFS（操作系统对文件的一种抽象）实现。

HDFS，是 Hadoop Distributed File System 的简称，是 Hadoop 抽象文件系统的一种实现。Hadoop 抽象文件系统可以与本地系统、Amazon S3 等集成，甚至可以通过 Web 协议（webhsfs）来操作。HDFS 的文件分布在集群机器上，同时提供副本进行容错及可靠性保证。例如客户端写入读取文件的直接操作都是分布在集群各个机器上的，没有单点性能压力。

# 2. HDFS 设计原则 #

HDFS 设计之初就非常明确其应用场景，适用与什么类型的应用，不适用什么应用，有一个相对明确的指导原则。

## 2.1 设计目标 ##

*  **存储非常大的文件**：这里非常大指的是几百 M、G、或者 TB 级别。实际应用中已有很多集群存储的数据达到 PB 级别。根据 Hadoop 官网，Yahoo！的 Hadoop 集群约有 10 万颗 CPU，运行在 4 万个机器节点上。更多世界上的 Hadoop 集群使用情况，参考 [Hadoop 官网][Hadoop].
 *  **采用流式的数据访问方式：**HDFS 基于这样的一个假设：最有效的数据处理模式是一次写入、多次读取数据集经常从数据源生成或者拷贝一次，然后在其上做很多分析工作   
    分析工作经常读取其中的大部分数据，即使不是全部。 因此读取整个数据集所需时间比读取第一条记录的延时更重要。
 *  **运行于商业硬件上**: Hadoop 不需要特别贵的、可靠性特别强的机器，可运行于普通商用机器（可以从多家供应商采购），商用机器不代表低端机器。在集群中（尤其是大的集群），节点失败率是比较高的 HDFS 的目标是确保集群在节点失败的时候不会让用户感觉到明显的中断。

## 2.2 HDFS 不适合的应用类型 ##

有些场景不适合使用 HDFS 来存储数据。下面列举几个：

1.  **低延时的数据访问。**对延时要求在毫秒级别的应用，不适合采用 HDFS。HDFS 是为高吞吐数据传输设计的, 因此可能牺牲延时 HBase 更适合低延时的数据访问。
2.  **大量小文件。**文件的元数据（如目录结构，文件 block 的节点列表，block-node mapping）保存在 NameNode 的内存中， 整个文件系统的文件数量会受限于 **NameNode 的内存大小**。 经验而言，一个文件 / 目录 / 文件块一般占有 150 字节的元数据内存空间。如果有 100 万个文件，每个文件占用 1 个文件块，则需要大约 300M 的内存。因此十亿级别的文件数量在现有商用机器上难以支持。
3.  **多方读写，需要任意的文件修改。**HDFS 采用追加（append-only）的方式写入数据。不支持文件任意 offset 的修改。不支持多个写入器（writer）。

# 3. HDFS 核心概念 #

##  ##

HDFS相比于其他的文件系统有什么特征呢?简单总结有如下的基本特征：

*  对于整个集群有单一的命名空间。
 *  数据一致性。适合一次写入多次读取的模型，客户端在文件没有被成功创建之前无法看到文件存在。
 *  文件会被分割成多个文件块，每个文件块被分配存储到数据节点上，而且根据配置会有复制文件块来保证数据的安全性。

在Hadoop中数据存储涉及HDFS的三个重要角色，分别为：名称节点(NameNode)、数据节点(DataNode)、客户端。

NameNode可以看做是分布式文件系统中的管理者，主要负责管理文件系统的命名空间、集群配置信息、存储块的复制。NameNode会存储文件系统的Metadata在内存中，这些信息主要包括文件信息，即每一个文件对应的文件块的信息，以及每一个文件块在DataNode的信息。

DataNode是文件存储的基本单元。它将Block存储在本地文件系统中，保存了Block的Metadata，同时周期性地发送所有存在的Block的报告给NameNode。Client就是需要获取分布式文件系统文件的应用程序。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzI2ODExMzc3_size_16_color_FFFFFF_t_70][]

**文件写入HDFS的基本流程如下：**

1.  Client向NameNode发起文件写入的请求。
2.  NameNode根据文件大小和文件块配置情况，向Client返回它所管理的DataNode的信息。
3.  Client将文件划分为多个Block，根据DataNode的地址信息，按顺序写入每一个DataNode中。

**文件读取HDFS的基本流程如下：**

1.  Client向NameNode发起文件读取的请求。
2.  NameNode返回文件存储的DataNode的信息。
3.  Client读取文件信息。

**在HDFS中复制文件块的基本流程如下：**

1.  NameNode发现部分文件的Block不符合最小复制数或部分DataNode失效。
2.  通知DataNode相互复制Block。
3.  DataNode开始相互复制。

## 3.1 Blocks ##

物理磁盘中有块的概念，磁盘的物理 Block 是磁盘操作最小的单元，读写操作均以 Block 为最小单元，一般为 512 Byte。文件系统在物理 Block 之上抽象了另一层概念，文件系统 Block 物理磁盘 Block 的整数倍。通常为几 KB。Hadoop 提供的 df、fsck 这类运维工具都是在文件系统的 Block 级别上进行操作。

**HDFS 的 Block 块比一般单机文件系统大得多，默认为 128M**。HDFS 的文件被拆分成 block-sized 的 chunk，chunk 作为独立单元存储。**比 Block 小的文件不会占用整个 Block，只会占据实际大小。例如， 如果一个文件大小为 1M，则在 HDFS 中只会占用 1M 的空间，而不是 128M。**

*  HDFS 的 Block 为什么这么大？

是为了最小化查找（seek）时间，控制定位文件与传输文件所用的时间比例。假设定位到 Block 所需的时间为 10ms，磁盘传输速度为 100M/s。如果要将定位到 Block 所用时间占传输时间的比例控制 1%，则 Block 大小需要约 100M。   
但是如果 Block 设置过大，在 MapReduce 任务中，Map 或者 Reduce 任务的个数 如果小于集群机器数量，会使得作业运行效率很低。

*  Block 抽象的好处

Block 的拆分使得单个文件大小可以大于整个磁盘的容量，构成文件的 Block 可以分布在整个集群， 理论上，单个文件可以占据集群中所有机器的磁盘。   
Block 的抽象也简化了存储系统，对于 Block，无需关注其权限，所有者等内容（这些内容都在文件级别上进行控制）。   
Block 作为容错和高可用机制中的副本单元，即以 Block 为单位进行复制。

## 3.2 Namenode & Datanode ##

整个 HDFS 集群由 Namenode 和 Datanode 构成 master-worker（主从）模式。Namenode 负责构建命名空间，管理文件的元数据等，而 Datanode 负责实际存储数据，负责读写工作。

### Namenode ###

Namenode 存放文件系统树及所有文件、目录的**元数据**。元数据持久化为 2 种形式：

*  namespcae image
 *  edit log

但是持久化数据中不包括 Block 所在的节点列表，及文件的 Block 分布在集群中的哪些节点上，这些信息在系统重启的时候重新构建（通过 Datanode 汇报的 Block 信息）。

在 HDFS 中，Namenode 可能成为集群的单点故障，Namenode 不可用时，整个文件系统是不可用的。HDFS 针对单点故障提供了 2 种解决机制：

1.  **备份持久化元数据。**将文件系统的元数据同时写到多个文件系统， 例如同时将元数据写到本地文件系统及 NFS。这些备份操作都是同步的、原子的。
2.  **Secondary Namenode。**Secondary 节点定期合并主 Namenode 的 namespace image 和 edit log， 避免 edit log 过大，通过创建检查点 checkpoint 来合并。它会维护一个合并后的 namespace image 副本， 可用于在 Namenode 完全崩溃时恢复数据。

下图为 Secondary Namenode 的管理界面：

![20160715204229964][]

Secondary Namenode 通常运行在另一台机器，因为合并操作需要耗费大量的 CPU 和内存。其数据落后于 Namenode，因此当 Namenode 完全崩溃时，会出现数据丢失。 通常做法是拷贝 NFS 中的备份元数据到 Second，将其作为新的主 Namenode。   
在 HA（High Availability 高可用性）中可以运行一个 Hot Standby，作为热备份，在 Active Namenode 故障之后，替代原有 Namenode 成为 Active Namenode。

### Datanode ###

数据节点负责存储和提取 Block，读写请求可能来自 namenode，也可能直接来自客户端。数据节点周期性向 Namenode 汇报自己节点上所存储的 Block 相关信息。

## 3.3 Block Caching（缓存块） ##

DataNode 通常直接从磁盘读取数据，但是频繁使用的 Block 可以在内存中缓存。**默认情况下，一个 Block 只有一个数据节点会缓存（**但是可以针对每个文件可以个性化配置）。   
作业调度器可以利用缓存提升性能，例如 MapReduce 可以把任务运行在有 Block 缓存的节点上。   
用户或者应用可以向 NameNode 发送缓存指令（缓存哪个文件，缓存多久）， 缓存池的概念用于管理一组缓存的权限和资源。

## 3.4 HDFS Federation ##

我们知道 **NameNode 的内存会制约文件数量**，HDFS Federation 提供了一种**横向扩展 NameNode 的方式**。在 Federation 模式中，每个 NameNode 管理命名空间的一部分，例如一个 NameNode 管理 / user 目录下的文件， 另一个 NameNode 管理 / share 目录下的文件。   
每个 NameNode 管理一个 namespace volumn，所有 volumn 构成文件系统的元数据。每个 NameNode 同时维护一个 Block Pool，保存 Block 的节点映射等信息。各 NameNode 之间是独立的，一个节点的失败不会导致其他节点管理的文件不可用。   
客户端使用 mount table 将文件路径映射到 NameNode。mount table 是在 Namenode 群组之上封装了一层，这一层也是一个 Hadoop 文件系统的实现，通过 viewfs: 协议访问。

## 3.5 HDFS HA(High Availability 高可用性) ##

在 HDFS 集群中，NameNode 依然是单点故障（SPOF: Single Point Of Failure）。**1）元数据同时写到多个文件系统以及 2）Second NameNode 定期 checkpoint** 有利于保护数据丢失，但是并不能提高可用性。   
这是因为 NameNode 是唯一一个对文件元数据和 file-block 映射负责的地方， 当它挂了之后，包括 MapReduce 在内的作业都无法进行读写。

当 NameNode 故障时，常规的做法是使用元数据备份重新启动一个 NameNode。元数据备份可能来源于：

*  多文件系统写入中的备份
 *  Second NameNode 的检查点文件

启动新的 Namenode 之后，需要重新配置客户端和 DataNode 的 NameNode 信息。另外重启耗时一般比较久，稍具规模的集群重启经常需要几十分钟甚至数小时，造成重启耗时的原因大致有：

1.  元数据镜像文件载入到内存耗时较长。 
2.  需要重放 edit log 
3.  需要收到来自 DataNode 的状态报告并且满足条件后才能离开安全模式提供写服务。

### Hadoop 的 HA 方案 ###

采用 HA 的 HDFS 集群配置两个 NameNode，分别处于 Active 和 Standby 状态。当 Active NameNode 故障之后，Standby 接过责任继续提供服务，用户没有明显的中断感觉。一般耗时在几十秒到数分钟。   
HA 涉及到的主要实现逻辑有

**1） 主备需共享 edit log 存储**  
主 NameNode 和待命的 NameNode 共享一份 edit log，当主备切换时，Standby 通过回放 edit log 同步数据。   
共享存储通常有 2 种选择

1.  NFS：传统的网络文件系统
2.  QJM：quorum journal manager

QJM 是专门为 HDFS 的 HA 实现而设计的，用来提供高可用的 edit log。QJM 运行一组 journal node，edit log 必须写到大部分的 journal nodes。通常使用 3 个节点，因此允许一个节点失败，类似 ZooKeeper。注意 QJM 没有使用 ZK，虽然 HDFS HA 的确使用了 ZK 来选举主 Namenode。一般推荐使用 QJM。

**2）DataNode 需要同时往主备发送 Block Report **  
因为 Block 映射数据存储在内存中（不是在磁盘上），为了在 Active NameNode 挂掉之后，新的 NameNode 能够快速启动，不需要等待来自 Datanode 的 Block Report，DataNode 需要同时向主备两个 NameNode 发送 Block Report。

**3）客户端需要配置 failover 模式（失效备援模式，对用户透明） **

Namenode 的切换对客户端来说是无感知的，通过客户端库来实现。客户端在配置文件中使用的 HDFS URI 是逻辑路径，映射到一对 Namenode 地址。客户端会不断尝试每一个 Namenode 地址直到成功。

**4）Standby 替代 Secondary NameNode **  
如果没有启用 HA，HDFS 独立运行一个守护进程作为 Secondary Namenode。定期 checkpoint，合并镜像文件和 edit 日志。

如果当主 Namenode 失败时，备份 Namenode 正在关机（停止 Standby），运维人员依然可以从头启动备份 Namenode，这样比没有 HA 的时候更省事，算是一种改进，因为重启整个过程已经标准化到 Hadoop 内部，无需运维进行复杂的切换操作。

NameNode 的切换通过代 failover controller 来实现。failover controller 有多种实现，默认实现使用 ZooKeeper 来保证只有一个 Namenode 处于 active 状态。

每个 Namenode 运行一个轻量级的 failover controller 进程，该进程使用简单的心跳机制来监控 Namenode 的存活状态并在 Namenode 失败时触发 failover。Failover 可以由运维手动触发，例如在日常维护中需要切换主 Namenode，这种情况 graceful(优雅的) failover，非手动触发的 failover 称为 ungraceful failover。

在 ungraceful failover 的情况下，没有办法确定失败（被判定为失败）的节点是否停止运行，也就是说触发 failover 后，之前的主 Namenode 可能还在运行。QJM 一次只允许一个 Namenode 写 edit log，但是之前的主 Namenode 仍然可以接受读请求。Hadoop 使用 fencing 来杀掉之前的 Namenode。Fencing 通过收回之前 Namenode 对共享的 edit log 的访问权限、关闭其网络端口使得原有的 Namenode 不能再继续接受服务请求。使用 STONITH 技术也可以将之前的主 Namenode 关机。

最后，HA 方案中 Namenode 的切换对客户端来说是不可见的，前面已经介绍过，主要通过客户端库来完成。

# 4. 命令行接口 #

HDFS 提供了各种交互方式，例如通过 Java API、HTTP、shell 命令行的。命令行的交互主要通过 hadoop fs 来操作。例如：

hadoop fs -copyFromLocal // 从本地复制文件到HDFS
    hadoop fs mkdir // 创建目录
    hadoop fs -ls  // 列出文件列表

Hadoop 中，文件和目录的权限类似于 POSIX 模型，包括读、写、执行 3 种权限：

1.  读权限（r）：用于读取文件或者列出目录中的内容
2.  写权限（w）：对于文件，就是文件的写权限。目录的写权限指在该目录下创建或者删除文件（目录）的权限。
3.  执行权限（x）：文件没有所谓的执行权限，被忽略。对于目录，执行权限用于访问器目录下的内容。

每个文件或目录都有 owner，group，mode 三个属性，owner 指文件的所有者，group 为权限组。mode   
由所有者权限、文件所属的组中组员的权限、非所有者非组员的权限组成。下图表示其所有者 root 拥有读写权限，supergroup 组的组员有读权限，其他人有读权限。

![20160715202602764][]

文件权限是否开启通过 dfs.permissions.enabled 属性来控制，这个属性默认为 false，没有打开安全限制，因此不会对客户端做授权校验，如果开启安全限制，会对操作文件的用户做权限校验。特殊用户 superuser 是 Namenode 进程的标识，不会针对该用户做权限校验。

最后看一下 ls 命令的执行结果：

![20160715202654359][]

这个返回结果类似于 Unix 系统下的 ls 命令，第一栏为文件的 mode，d 表示目录，紧接着 3 种权限 9 位。 第二栏是指文件的副本数，这个数量通过`dfs.replication`配置，目录则使用 - 表示没有副本一说。其他诸如所有者、组、更新时间、文件大小跟 Unix 系统中的 ls 命令一致。

如果需要查看集群状态或者浏览文件目录，可以访问 Namenode 暴露的 Http Server 查看集群信息，一般在 namenode 所在机器的 50070 端口。

![20160715223051523][]

![20160715223105739][]

![20160715223118601][]

## 5. Hadoop 文件系统 ##

前面 Hadoop 的文件系统概念是抽象的，HDFS 只是其中的一种实现。Hadoop 提供的实现如下图：

![20160715204510171][]

简单介绍一下，Local 是对本地文件系统的抽象；HDFS 就是我们最常见的，两种 web 形式（WebHDFS，SecureWebHDFS）的实现通过 HTTP 提供文件操作接口；HAR 是 Hadoop 体系下的压缩文件，当文件很多的时候可以压缩成一个大文件，可以有效减少元数据的数量。

![20160715204527148][]

Viewfs 就是我们前面介绍 HDFS Federation 张提到的，用来在客户端屏蔽多个 Namenode 的底层细节；FTP 顾名思义，就是使用 ftp 协议来实现，对文件的操作转化为 ftp 协议；S3A 是对 Amazon 云服务提供的存储系统的实现；Azure 则是微软的云服务平台实现；Swift是Openstack的云服务平台实现。

前面我们提到了使用命令行跟 HDFS 交互，事实上还有很多方式来操作文件系统。例如 Java 应用程序可以使用 org.apache.hadoop.fs.FileSystem 来操作，其他形式的操作也都是基于 FileSystem 进行封装。我们这里主要介绍一下 HTTP 的交互方式。   
WebHDFS 和 SWebHDFS 协议将文件系统暴露 HTTP 操作，这种交互方式比原生的 Java 客户端慢，不适合操作大文件。通过 HTTP，有 2 种访问方式，直接访问和通过代理访问

**1. 直接访问 **

![20160715205452074][]

Namenode 和 Datanode 默认打开了嵌入式 web server，即 dfs.webhdfs.enabled 默认为 true。webhdfs 通过这些服务器来交互。元数据的操作通过 namenode 完成，文件的读写首先发到 namenode，然后重定向到 datanode 读取（写入）实际的数据流。

**2. 通过 HDFS 代理访问**

![20160715205504590][]

采用代理的示意图如上所示。 使用代理的好处是可以通过代理实现负载均衡或者对带宽进行限制，或者防火墙设置。代理通过 HTTP 或者 HTTPS 暴露为 WebHDFS，对应为 webhdfs 和 swebhdfs URL Schema。

代理作为独立的守护进程，独立于 namenode 和 datanode，使用 httpfs.sh 脚本，默认运行在 14000 端口

除了 FileSystem 直接操作，命令行，HTTTP 外，还有 C 语言 API，NFS，FUSER 等方式，这里不做过多介绍。

# 6. Java 接口 #

实际的应用中，对 HDFS 的大多数操作还是通过 FileSystem 来操作，这部分重点介绍一下相关的接口，主要关注 HDFS 的实现类 DistributedFileSystem 及相关类。

## 6.1 读操作 ##

可以使用 URL 来读取数据，或者直接使用 FileSystem 操作。

### 从 Hadoop URL 读取数据 ###

java.net.URL 类提供了资源定位的统一抽象，任何人都可以自己定义一种 URL Schema，并提供相应的处理类来进行实际的操作。hdfs schema 便是这样的一种实现。

InputStream in = null;
    try {
     in = new URL("hdfs://master/user/hadoop").openStream();
    }finally{
     IOUtils.closeStream(in);
    }

为了使用自定义的 Schema，需要设置 URLStreamHandlerFactory，这个操作一个 JVM 只能进行一次，多次操作会导致不可用，通常在静态块中完成。下面的截图是一个使用示例：

![20160715210623079][]

![20160715210635185][]

### 使用 FileSystem API 读取数据 ###

1） 首先获取 FileSystem 实例，一般使用静态 get 方法

public static FileSystem get(Configuration conf) throws IOException
    public static FileSystem get(URI uri , Configuration conf) throws IOException
    public static FileSystem get(URI uri , Configuration conf，String user) throws IOException

如果是本地文件，通过 getLocal 获取本地文件系统对象：

2）调用 FileSystem 的 open 方法获取一个输入流:

public FSDataInputStream open(Path f) throws IOException
    public abstarct FSDataInputStream open(Path f , int bufferSize) throws IOException

默认情况下，open 使用 4KB 的 Buffer，可以根据需要自行设置。

3）使用 FSDataInputStream 进行数据操作   
FSDataInputStream 是 java.io.DataInputStream 的特殊实现，在其基础上增加了随机读取、部分读取的能力

public class FSDataInputStream extends DataInputStream
        implements Seekable, PositionedReadable, 
          ByteBufferReadable, HasFileDescriptor, CanSetDropBehind, CanSetReadahead,
          HasEnhancedByteBufferAccess

随机读取操作通过 Seekable 接口定义：

public interface Seekable {
        void seek(long pos) throws IOException;
        long getPos() throws IOException;
    }

seek 操作开销昂贵，慎用。

部分读取通过 PositionedReadable 接口定义：

public interface PositionedReadable{
        public int read(long pistion ,byte[] buffer,int offser , int length) throws IOException;
        public int readFully(long pistion ,byte[] buffer,int offser , int length) throws IOException;
        public int readFully(long pistion ,byte[] buffer) throws IOException;
    }

## 6.2 写数据 ##

在 HDFS 中，文件使用 FileSystem 类的 create 方法及其重载形式来创建，create 方法返回一个输出流 FSDataOutputStream，可以调用返回输出流的 getPos 方法查看当前文件的位移，但是不能进行 seek 操作，HDFS 仅支持追加操作。

创建时，可以传递一个回调接口 Peofressable，获取进度信息

append(Path f) 方法用于追加内容到已有文件，但是并不是所有的实现都提供该方法，例如 Amazon 的文件实现就没有提供追加功能。

下面是一个例子：

String localSrc =  args[0];
    String dst = args[1];
     
    InputStream in = new BufferedInputStream(new FileInputStream(localSrc));
     
    Configuration conf = new Configuration();
    FileSystem fs = FileSystem.get(URI.create(dst),conf);
     
    OutputStream out = fs.create(new Path(dst), new Progressable(){
        public vid progress(){
            System.out.print(.);
        }
    });
     
    IOUtils.copyBytes(in , out, 4096,true);

## 6.3 目录操作 ##

使用 mkdirs（）方法, 会自动创建没有的上级目录

HDFS 中元数据封装在 FileStatus 类中，包括长度、block size，replicaions，修改时间、所有者、权限等信息。使用 FileSystem 提供的 getFileStatus 方法获取 FileStatus。exists() 方法判断文件或者目录是否存在；

列出文件（list），则使用 listStatus 方法，可以查看文件或者目录的信息

public abstract FileStatus[] listStatus(Path f) throws FileNotFoundException, 
                                                             IOException;

Path 是个文件的时候，返回长度为 1 的数组。FileUtil 提供的 stat2Paths 方法用于将 FileStatus 转化为 Path 对象。

globStatus 则使用通配符对文件路径进行匹配：

PathFilter 用于自定义文件名过滤，不能根据文件属性进行过滤，类似于 java.io.FileFilter。例如下面这个例子排除到给定正则表达式的文件：

public interfacePathFilter{
        boolean accept(Path path);
    }

## 6.4 删除数据 ##

使用 FileSystem 的 delete() 方法

recursive 
    参数在 f 是个文件的时候被忽略。如果 f 是文件并且 recursice 为 true，则删除整个目录，否则抛出异常.

# 7. 数据流 (读写流程） #

接下来详细介绍 HDFS 读写数据的流程，以及一致性模型相关的一些概念。

## 7.1 读文件 ##

大致读文件的流程如下：

![20160715215428741][]

1.  客户端传递一个文件 Path 给 FileSystem 的 open 方法
2.  DFS 采用 RPC 远程获取文件最开始的几个 block 的 datanode 地址。Namenode 会根据网络拓扑结构决定返回哪些节点（前提是节点有 block 副本），如果客户端本身是 Datanode 并且节点上刚好有 block 副本，直接从本地读取。
3.  客户端使用 open 方法返回的 FSDataInputStream 对象读取数据（调用 read 方法）
4.  DFSInputStream（FSDataInputStream 实现了改类）连接持有第一个 block 的、最近的节点，反复调用 read 方法读取数据
5.  第一个 block 读取完毕之后，寻找下一个 block 的最佳 datanode，读取数据。如果有必要，DFSInputStream 会联系 Namenode 获取下一批 Block 的节点信息 (存放于内存，不持久化），这些寻址过程对客户端都是不可见的。
6.  数据读取完毕，客户端调用 close 方法关闭流对象

在读数据过程中，如果与 Datanode 的通信发生错误，DFSInputStream 对象会尝试从下一个最佳节点读取数据，并且记住该失败节点， 后续 Block 的读取不会再连接该节点   
读取一个 Block 之后，DFSInputStram 会进行检验和验证，如果 Block 损坏，尝试从其他节点读取数据，并且将损坏的 block 汇报给 Namenode。   
客户端连接哪个 datanode 获取数据，是由 namenode 来指导的，这样可以支持大量并发的客户端请求，namenode 尽可能将流量均匀分布到整个集群。   
Block 的位置信息是存储在 namenode 的内存中，因此相应位置请求非常高效，不会成为瓶颈。

## 7.2 写文件 ##

![20160715215729850][]

1.  客户端调用 DistributedFileSystem 的 create 方法
2.  DistributedFileSystem 远程 RPC 调用 Namenode 在文件系统的命名空间中创建一个新文件，此时该文件没有关联到任何 block。 这个过程中，Namenode 会做很多校验工作，例如是否已经存在同名文件，是否有权限，如果验证通过，返回一个 FSDataOutputStream 对象。 如果验证不通过，抛出异常到客户端。
3.  客户端写入数据的时候，DFSOutputStream 分解为 packets（数据包），并写入到一个数据队列中，该队列由 DataStreamer 消费。
4.  DateStreamer 负责请求 Namenode 分配新的 block 存放的数据节点。这些节点存放同一个 Block 的副本，构成一个管道。 DataStreamer 将 packet 写入到管道的第一个节点，第一个节点存放好 packet 之后，转发给下一个节点，下一个节点存放 之后继续往下传递。
5.  DFSOutputStream 同时维护一个 ack queue 队列，等待来自 datanode 确认消息。当管道上的所有 datanode 都确认之后，packet 从 ack 队列中移除。
6.  数据写入完毕，客户端 close 输出流。将所有的 packet 刷新到管道中，然后安心等待来自 datanode 的确认消息。全部得到确认之后告知 Namenode 文件是完整的。 Namenode 此时已经知道文件的所有 Block 信息（因为 DataStreamer 是请求 Namenode 分配 block 的），只需等待达到最小副本数要求，然后返回成功信息给客户端。

Namenode 如何决定副本存在哪个 Datanode？

HDFS 的副本的存放策略是**可靠性、写带宽、读带宽**之间的权衡。默认策略如下：

*  第一个副本放在客户端相同的机器上，如果机器在集群之外，随机选择一个（但是会尽可能选择容量不是太慢或者当前操作太繁忙的）
 *  第二个副本随机放在不同于第一个副本的机架上。
 *  第三个副本放在跟第二个副本同一机架上，但是不同的节点上，满足条件的节点中随机选择。
 *  更多的副本在整个集群上随机选择，虽然会尽量避免太多副本在同一机架上。   
    副本的位置确定之后，在建立写入管道的时候，会考虑网络拓扑结构。下面是可能的一个存放策略：

![20160715220056463][]

这样选择很好滴平衡了可靠性、读写性能

*  可靠性：Block 分布在两个机架上
 *  写带宽：写入管道的过程只需要跨越一个交换机
 *  读带宽：可以从两个机架中任选一个读取

## 7.3 一致性模型 ##

一致性模型描述文件系统中读写操纵的可见性。HDFS 中，文件一旦创建之后，在文件系统的命名空间中可见：

Path p = new Path("p");
    fs.create(p);
    assertTaht(fs.exists(p),is(true));

但是任何被写入到文件的内容不保证可见，即使对象流已经被刷新。

“`java 
    Path p = new Path(“p”); 
    OutputStream out = fs.create(p); 
    out.write(“content”.getBytes(“UTF-8”)); 
    out.flush(); 
    assertTaht(fs.getFileStatus(p).getLen,0L); // 为 0，即使调用了 flush
    
    如果需要强制刷新数据到Datanode，使用FSDataOutputStream的hflush方法强制将缓冲刷到datanode
    hflush之后，HDFS保证到这个时间点为止写入到文件的数据都到达所有的数据节点。
     ```java
    Path p = new Path("p");
    OutputStream out = fs.create(p);
    out.write("content".getBytes("UTF-8"));
    out.flush();
    assertTaht(fs.getFileStatus(p).getLen,is(((long,"content".length())));

关闭对象流时，内部会调用 hflush 方法, 但是 hflush 不保证 datanode 数据已经写入到磁盘，只是保证写入到 datanode 的内存， 因此在机器断电的时候可能导致数据丢失，如果要保证写入磁盘，使用 hsync 方法，hsync 类型与 fsync（）的系统调用，fsync 提交某个文件句柄的缓冲数据。

FileOutputStreamout = new FileOutPutStream(localFile);
    out.write("content".getBytes("UTF-8"));
    out.flush();
    out.getFD().sync();
    assertTaht(localFile.getLen,is(((long,"content".length())));

使用 hflush 或 hsync 会导致吞吐量下降，因此设计应用时，需要在吞吐量以及数据的健壮性之间做权衡。

另外，文件写入过程中，当前正在写入的 Block 对其他 Reader 不可见。

## 7.4 Hadoop 节点距离 ##

在读取和写入的过程中，namenode 在分配 Datanode 的时候，会考虑节点之间的距离。**HDFS 中，距离没有   
采用带宽来衡量**，因为实际中很难准确度量两台机器之间的带宽。   
**Hadoop 把机器之间的拓扑结构组织成树结构，并且用到达公共父节点所需跳转数之和作为距离**。事实上这是一个距离矩阵的例子。下面的例子简明地说明了距离的计算：

![20160715221045921][]

*  同一数据中心，同一机架，同一节点距离为 0
 *  同一数据中心，同一机架，不同节点距离为 2
 *  同一数据中心，不同机架，不同节点距离为 4
 *  不同数据中心，不同机架，不同节点距离为 6

Hadoop 集群的拓扑结构需要手动配置，如果没配置，Hadoop 默认所有节点位于同一个数据中心的同一机架上。

# 8. 相关运维工具 #

## 8.1 使用 distcp 并行复制 ##

前面的关注点都在于单线程的访问，如果需要并行处理文件，需要自己编写应用。Hadoop 提供的 distcp 工具用于并行导入数据到 Hadoop 或者从 Hadoop 导出。一些例子：

hadoop distcp file1 file2  //可以作为fs -cp命令的高效替代
    hadoop distcp dir1 dir2
    hadoop distcp -update dir1 dir2 #update参数表示只同步被更新的文件，其他保持不变

distcp 是底层使用 MapReduce 实现，只有 map 实现，没有 reduce。在 map 中并行复制文件。 distcp 尽可能在 map 之间平均分配文件。map 的数量可以通过 - m 参数指定:

这样的操作常用于在两个集群之间复制数据，update 参数表示只同步被更新过的数据，delete 会删除目标目录中存在，但是源目录不存在的文件。p 参数表示保留文件的全校、block 大小、副本数量等属性。

如果两个集群的 Hadoop 版本不兼容，可以使用 webhdfs 协议：

## 8.2 平衡 HDFS 集群 ##

在 distcp 工具中，如果我们指定 map 数量为 1，不仅速度很慢，每个 Block 第一个副本将全部落到运行这个唯一 map 的节点上，直到磁盘溢出。因此使用 distcp 的时候，最好使用默认的 map 数量，即 20.   
HDFS 在 Block 均匀分布在各个节点上的时候工作得最好，如果没有办法在作业中尽量保持集群平衡，例如为了限制 map 数量（以便其他节点可以被别的作业使用），那么可以使用 balancer 工具来调整集群的 Block 分布。

# 参考 #

主要参考《Hadoop》权威指南第 3 章，自己进一步整理。感谢原作者提供这么好的书籍。

[20160715085646356]: /images/20230531/bb938de84cd744e380b671ae5b721000.png
[20160715085700575]: /images/20230531/6a5f9849f18f4720a6289b5bcf4f6bb8.png
[Hadoop]: https://wiki.apache.org/hadoop/PoweredBy
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzI2ODExMzc3_size_16_color_FFFFFF_t_70]: /images/20230531/481c4d994bed4f7e8a738abb12a0f32d.png
[20160715204229964]: /images/20230531/cf2b1c80b34a44e4aff181d5ebdc0f64.png
[20160715202602764]: /images/20230531/6579607c55e74fd895691efb9d4ac893.png
[20160715202654359]: /images/20230531/4b89f00fa0194e5294e4f04e3de321e5.png
[20160715223051523]: /images/20230531/a363aa68178045108deef4ee7c9d326e.png
[20160715223105739]: /images/20230531/8ea4a361d09749d785d498b586eb1ae4.png
[20160715223118601]: /images/20230531/e08f48c00ca7417cb548a41fb3c0f587.png
[20160715204510171]: /images/20230531/ae614462c9fe49438a5530698b705c49.png
[20160715204527148]: /images/20230531/e47bddfa61ee4095a5d857e1516f9648.png
[20160715205452074]: /images/20230531/27d5784b80b5473d84655cf7ae41a2d0.png
[20160715205504590]: /images/20230531/97fe95a5424b4162b832bdf566bd5076.png
[20160715210623079]: /images/20230531/1fbb2b3ac2c74d0e9a425d65289a0957.png
[20160715210635185]: /images/20230531/62a57b711acb4aa09e12edaa988fb728.png
[20160715215428741]: /images/20230531/1982e8b20693499283391d979692ba0a.png
[20160715215729850]: /images/20230531/14faf3e0e58249f08d5a0000fb04ea10.png
[20160715220056463]: /images/20230531/f2d74b607831494f96d8e44a1d67e206.png
[20160715221045921]: /images/20230531/ef532c8d844644df91f375a7f5fad696.png