大数据存储域——HDFS存储系统

摘要

本文介绍了HDFS存储系统，包括其组件、工作机制、实战经验总结、使用场景以及与SpringBoot的实战示例和优化设计。HDFS由Client、NameNode、SecondaryNameNode、DataNode等组件构成，通过特定的工作机制实现文件的读取和写入。它适用于多种场景，如日志采集、大数据离线分析等，但也有不适用的场景。文中还展示了如何在SpringBoot项目中使用HDFS，包括引入依赖、配置文件、配置类、操作服务类和控制器层的实现。最后探讨了HDFS的优化设计。

1. HDFS分布式存储系统简介

Apache Hadoop是一个开源的分布式计算框架，专门用来处理大规模数据集。Hadoop 是一套分布式系统框架，包括：分布式存储技术 HDFS、分布式资源调度技术 YARN、分布式计算框架 MapReduce。

• HDFS： 是 Hadoop 的基础存储系统，Master是NameNode，Slave是DataNode。
• YARN： 是从 Hadoop 2.0 开始引入的新资源管理框架，主要包括 ResourceManager、NodeManager。
• MapReduce： 是最早期的 Hadoop 计算模型，现在很多场景已被 Spark、Flink 替代（效率更高，支持流处理）。

除了上述核心，Hadoop 还衍生出一系列生态工具：

1.1. HDFS的组件

HDFS (Hadoop Distributed File System, 分布式文件系统) 是Google公司的GFS论文思想的实现，也作为 Hadoop 的存储系统，它包含客户端(Client)、元数据节点(NameNode)、备份节点(Secondary NameNode)以及数据存储节点(DataNode)。

1.1.1. Client

HDFS 利用分布式集群节点来存储数据，并提供统一的文件系统访问接口。这样，用户在使用分布式文件系统时就如同在使用普通的单节点文件系统一样，仅通过对 NameNode 进行交互访问就可以实现操作HDFS中的文件。HDFS提供了非常多的客户端，包括命令行接口、Java API、Thrift接口、Web界面等。

1.1.2. NameNode

NameNode 作为 HDFS 的管理节点，负责保存和管理分布式系统中所有文件的元数据信息，如果将 HDFS 比作一本书，那么 NameNode 可以理解为这本书的目录。

其职责主要有以下三点：

1. 负责接收 Client 发送过来的读写请求；
2. 管理和维护HDFS的命名空间: 元数据是以镜像文件(fsimage)和编辑日志(editlog)两种形式存放在本地磁盘上的，可以记录 Client 对 HDFS 的各种操作，比如修改时间、访问时间、数据块信息等。
3. 监控和管理DataNode：负责监控集群中DataNode的健康状态，一旦发现某个DataNode宕掉，则将该 DataNode 从 HDFS 集群移除并在其他 DataNode 上重新备份该 DataNode 的数据(该过程被称为数据重平衡，即rebalance)，以保障数据副本的完整性和集群的高可用性。

1.1.3. SecondaryNameNode

SecondaryNameNode 是 NameNode 元数据的备份，在NameNode宕机后，SecondaryNameNode 会接替 NameNode 的工作，负责整个集群的管理。并且出于可靠性考虑，SecondaryNameNode 节点与 NameNode 节点运行在不同的机器上，且 SecondaryNameNode 节点与 NameNode 节点的内存要一样大。

同时，为了减小 NameNode 的压力，NameNode 并不会自动合并 HDFS中的元数据镜像文件(fsimage)和编辑日志(editlog)，而是将该任务交由 SecondaryNameNode 来完成，在合并完成后将结果发送到NameNode, 并再将合并后的结果存储到本地磁盘。

1.1.4. DataNode

存放在HDFS上的文件是由数据块组成的，所有这些块都存储在DataNode节点上。DataNode 负责具体的数据存储，并将数据的元信息定期汇报给 NameNode，并在 NameNode 的指导下完成数据的 I/O 操作。

实际上，在DataNode节点上，数据块就是一个普通文件，可以在DataNode存储块的对应目录下看到（默认在$(dfs.data.dir)/current的子目录下），块的名称是 blk_ID，其大小可以通过dfs.blocksize设置，默认为128MB。

初始化时，集群中的每个 DataNode 会将本节点当前存储的块信息以块报告的形式汇报给 NameNode。在集群正常工作时，DataNode 仍然会定期地把最新的块信息汇报给 NameNode，同时接收 NameNode 的指令，比如创建、移动或删除本地磁盘上的数据块等操作。

1.1.5. HDFS数据副本

HDFS 文件系统在设计之初就充分考虑到了容错问题，会将同一个数据块对应的数据副本（副本个数可设置，默认为3）存放在多个不同的 DataNode 上。在某个 DataNode 节点宕机后，HDFS 会从备份的节点上读取数据，这种容错性机制能够很好地实现即使节点故障而数据不会丢失。

1.2. HDFS的工作机制

1.2.1. NameNode 工作机制

NameNode简称NN

1. NN 启动后,会将镜像文件(fsimage)和编辑日志(editlog)加载进内存中；
2. 客户端发来增删改查等操作的请求；
3. NN 会记录下操作，并滚动日志，然后在内存中对操作进行处理。

1.2.2. SecondaryNameNode工作机制

SecondaryNameNode简称2NN

1. 当编辑日志数据达到一定量或者每隔一定时间，就会触发 2NN 向 NN 发出 checkpoint请求；
2. 如果发出的请求有回应，2NN 将会请求执行 checkpoint 请求；
3. 2NN 会引导 NN 滚动更新编辑日志，并将编辑日志复制到 2NN 中；
4. 同编辑日志一样，将镜像文件复制到 2NN 本地的 checkpoint 目录中；
5. 2NN 将镜像文件导入内存中，回放编辑日志，将其合并到新的fsimage.ckpt；
6. 将 fsimage.ckpt 压缩后写入到本地磁盘；
7. 2NN 将 fsimage.ckpt 传给 NN；
8. NN 会将新的 fsimage.ckpt 文件替换掉原来的 fsimage，然后直接加载和启用该文件。

1.2.3. HDFS文件的读取流程

1. 客户端调用FileSystem 对象的open()方法，其实获取的是一个分布式文件系统（DistributedFileSystem）实例；
2. 将所要读取文件的请求发送给 NameNode，然后 NameNode 返回文件数据块所在的 DataNode 列表（是按照 Client 距离 DataNode 网络拓扑的远近进行排序的），同时也会返回一个文件系统数据输入流（FSDataInputStream）对象；
3. 客户端调用 read() 方法，会找出最近的 DataNode 并连接；
4. 数据从 DataNode 源源不断地流向客户端。

1.2.4. HDFS文件的写入流程

1. 客户端通过调用分布式文件系统（DistributedFileSystem）的create()方法创建新文件；
2. DistributedFileSystem 将文件写入请求发送给 NameNode，此时 NameNode 会做各种校验，比如文件是否存在，客户端有无权限去创建等；
3. 如果校验不通过则会抛出I/O异常。如果校验通过，NameNode 会将该操作写入到编辑日志中，并返回一个可写入的 DataNode 列表，同时，也会返回文件系统数据输出流（FSDataOutputStream）的对象；
4. 客户端在收到可写入列表之后，会调用 write() 方法将文件切分为固定大小的数据包，并排成数据队列；
5. 数据队列中的数据包会写入到第一个 DataNode，然后第一个 DataNode 会将数据包发送给第二个 DataNode，依此类推。
6. DataNode 收到数据后会返回确认信息，等收到所有 DataNode 的确认信息之后，写入操作完成。

2. HDFS实战经验总结

2.1. HDFS存储系统和Mysql数据库有什么区别？

这是一个非常经典的问题，HDFS 和 MySQL 都可以“存数据”，HDFS 是为了解决“海量数据的高吞吐分布式存储”，MySQL 是为了解决“结构化数据的实时存储与查询”问题。但它们的定位、设计目标、使用场景完全不同。下面我们从多个维度来对比，帮你全面理解它们的区别。

2.1.1. 🆚 HDFSvsMySQL：核心区别

2.1.2. 🧠 HDFS 和 MySQL比较

2.1.3. 📌 举个具体例子：

假设你是风控系统的开发者：

• 你用MySQL存：用户表、借款记录表、规则配置表，支持事务操作和实时查询。
• 你用HDFS存：用户调用记录日志、历史交易大数据、用户行为日志、模型训练数据等，支持后续批量分析。

2.2. HDFS存储系统的使用场景？

HDFS（Hadoop Distributed File System）是专门为大数据存储与处理设计的分布式文件系统。它不是用来替代数据库的，而是为了解决传统存储系统难以处理海量数据、超大文件、高吞吐量、分布式读写的问题。以下是 HDFS 最常见的典型使用场景，你可以根据业务系统的特点对号入座：

2.2.1. 日志采集与分析

📁 应用场景：收集海量系统日志、用户行为日志、访问记录、接口调用日志等。

📌 为什么适合 HDFS：

• 数据量大（PB 级别）
• 文件多且持续产生（写一次、读多次）
• 不需要实时查询，适合离线分析

✅ 常与 Flume + Hive/Spark 组合使用。

2.2.2. 大数据离线分析

📁 应用场景：大数据报表、风控画像、模型训练、批量数据清洗等。

📌 为什么适合 HDFS：

• 存储 TB / PB 级原始数据或中间计算结果
• 结合 Hive、Spark、MapReduce 实现大规模分析任务

✅ 适用于风控系统中的用户信用评分建模、交易异常批量分析等场景。

2.2.3. 机器学习/模型训练数据存储

📁 应用场景：用于训练风控模型、推荐系统、CTR模型等的训练数据。

📌 为什么适合 HDFS：

• 支持大文件、批量处理
• 与 Spark ML、TensorFlow on Hadoop 等兼容

✅ 可结合 Spark MLlib、TensorFlow on HDFS 使用。

2.2.4. 视频、图片等大文件归档

📁 应用场景：视频监控数据、图像识别数据、大量媒体内容存储。

📌 为什么适合 HDFS：

• 文件大（几百 MB 至 GB）
• 不需要高并发小文件随机读写
• 存储成本低，可用廉价硬盘搭建

✅ 适合“冷数据”存档、后续批处理。

2.2.5. 数据中台/湖仓一体系统的底层存储

📁 应用场景：大数据中台、数据湖架构中用于统一存储各类结构化、非结构化数据。

📌 为什么适合 HDFS：

• • 与 Hive、Iceberg、Hudi 等表格式存储系统集成
  • 支持数据分区、Schema 管理、演化

✅ 可作为 Data Lake（数据湖）中的底层文件系统。

2.2.6. ETL/数据清洗任务的中间文件存储

📁 应用场景：在数据从业务系统导入、加工、清洗的过程中，作为中间结果文件存储系统。

📌 为什么适合 HDFS：

• 高吞吐读写
• 支持批量清洗、转换任务（如 Spark Job）

✅ 通常搭配 Spark、Hive、Oozie 使用。

2.2.7. 企业级风控/征信系统中的大数据底座

📁 应用场景：风控评分引擎、规则引擎的大数据支撑层。

📌 为什么适合 HDFS：

• 存储来自多渠道的数据源（APP、WEB、第三方接口）
• 支持跨平台、跨语言的访问（Java、Python、Scala）

2.2.8. 🚫 不适合 HDFS 的场景（谨记）

2.3. HDFS 中NameNode 是怎么管理dataNode，他们是怎么通信的？

在 HDFS（Hadoop Distributed File System）中，NameNode 与 DataNode 之间的通信机制是 HDFS 架构中最核心的部分之一。它实现了文件系统的元数据管理和数据块的实际存储的协作。NameNode 负责大脑调度与指挥（元数据、分配 block、恢复副本），DataNode 负责干活（存储数据块）。二者之间通过 心跳、Block 报告、命令下发实现协调合作。 下面详细说明：

2.3.1. 🧠 NameNode 与 DataNode 的职责

2.3.2. 🔌 通信机制详解

HDFS 中 NameNode 和 DataNode 是通过TCP 协议进行通信的，主要有两种类型的通信：

2.3.2.1. 1️⃣ 心跳（Heartbeat）

• 目的：让 NameNode 知道哪些 DataNode 还活着；
• 频率：每 3 秒一次（默认 dfs.heartbeat.interval=3）；
• 通道：DataNode 主动向 NameNode 发送；
• 携带内容：

• • DataNode 的 ID；
  • 当前使用量、剩余空间；
  • 是否处于运行状态；

• NameNode 响应：

• • 是否需要发送 block 报告（Block Report）；
  • 是否需要复制、删除某些 block；
  • 启动 block 的复制或删除命令；

2.3.2.2. 2️⃣ Block 报告（Block Report）

• 目的：告诉 NameNode：这个 DataNode 存储了哪些 block；
• 频率：每 6 小时（默认 dfs.blockreport.intervalMsec=21600000）；
• 内容：所有 block 的 ID、长度、版本等；
• 作用：维护 NameNode 的 block-to-DataNode 映射；

2.3.2.3. 3️⃣ 增量 block 报告（Incremental Block Report）

• 触发时机：当 DataNode 有新的 block 写入、删除或复制；
• 作用：比 Block 报告更及时，减少元数据滞后；

2.3.2.4. 4️⃣ 客户端读写时的协调（NameNode 作为调度者）

• 写操作：

• • 客户端向 NameNode 请求写入；
  • NameNode 返回可用的若干 DataNode 地址（用于副本）；
  • 客户端直接与这些 DataNode 通信写入 block；

• 读操作：

• • 客户端向 NameNode 请求读取文件；
  • NameNode 返回 block 所在的 DataNode 列表；
  • 客户端直接从 DataNode 拉取数据；

2.3.3. 🔄 容错与故障检测机制

• 超时判定失效节点：

• • 如果某个 DataNode 在 10.5 分钟内（默认 3 秒 × 210 次）没有心跳，NameNode 会判定它挂掉（dfs.namenode.heartbeat.recheck-interval）；

• 副本恢复：

• • NameNode 会自动触发 block 复制以确保副本数量（dfs.replication）；
  • 可通过其他 DataNode 重新恢复副本到健康节点；

2.4. 🧪 可视化示意图

[Client]|                     写/读请求v
[NameNode] <--------- 心跳、Block报告 --------- [DataNode 1]|                                        [DataNode 2]|                                        [DataNode 3]|响应 block 位置信息|
[Client 与 DataNode 通信完成真正的 I/O]

2.5. HDFS中NameNode和DataNode是通过TCP协议进行通信的是通过发送http请求吗？

HDFS中NameNode和DataNode 是通过 TCP 协议通信的，但不是通过 HTTP 请求通信，主要通过以下两种机制：

2.5.1. ✅ 通信协议：HDFS自定义的RPC（远程过程调用）协议

• NameNode 与 DataNode、Client 之间的通信使用的是 Hadoop 自定义的 RPC 协议，底层是基于 TCP。
• 它不是 HTTP 协议，而是 通过Java 的 Netty/NIO 框架 自定义的高性能传输机制。

2.5.2. ✅ NameNode与DataNode通信流程

2.5.2.1. 📌 启动时注册

• DataNode 启动后，会向 NameNode 发起注册请求（RPC 形式），发送自己的 IP、端口、存储容量等信息。

2.5.2.2. 📌 定期心跳（默认3秒）

• DataNode 每隔一段时间（默认 3 秒）向 NameNode 发送心跳包，报告存活状态。
• 如果 NameNode 超过 10.5 秒（默认值）没收到心跳，就认为该节点挂了。

2.5.2.3. 📌 Block 汇报

• 每隔一小时或在 Block 变更时，DataNode 会向 NameNode 发送 block 汇报，说明自己当前有哪些 block。

2.5.3. ✅ Client与DataNode的通信

• 当客户端（Client）读写数据时，会先通过 RPC 向 NameNode 获取数据块位置（block 和 DataNode 的映射信息）。
• 然后客户端通过 TCP 直接和 DataNode 通信来读写数据，这时候 使用的是 HTTP 协议的子集（叫做 DataTransferProtocol），仍然是基于 TCP。

2.5.4. 🔁 NameNode和DataNode总结

2.6. DataNode 每隔一段时间（默认 3 秒）向 NameNode 发送心跳包，报告存活状态。 是通过定时任务触发吗？

是的，DataNode 向 NameNode 发送心跳包是通过定时任务触发的，具体实现是在 DataNode 的内部线程机制中完成的。

2.6.1. 🔧 DataNode 心跳机制的执行原理

2.6.1.1. ✅ 背景线程：BPServiceActor

DataNode 启动时，会为每一个 Block Pool 启动一个 BPServiceActor（Block Pool Service Actor）线程，这个线程的主要任务包括：

• 定期向 NameNode 发送心跳（Heartbeat）
• 汇报 block 列表（Block Report）
• 发送增量的 block 通知（Incremental Block Report）

这些任务都通过定时机制触发。

2.6.1.2. ✅ 心跳定时任务调度逻辑

在源码中，心跳逻辑大致如下（Hadoop 3.x）：

// 每个 BPServiceActor 都维护一个线程，周期性执行：
while (shouldRun) {long startTime = Time.monotonicNow();// 心跳间隔检查if (shouldSendHeartbeat()) {sendHeartBeat(); // 发送心跳}// 处理 Block Report、Command 等等processCommandFromNameNode();long sleepTime = computeNextWaitTime(startTime);Thread.sleep(sleepTime);
}

默认心跳间隔配置项：dfs.heartbeat.interval，默认值是 3 秒。

2.7. 🔍 NameNode如何处理心跳

NameNode 接收到心跳包后，会：

1. 更新对应 DataNode 的“最后心跳时间”
2. 检查 DataNode 的状态（健康、存储、负载等）
3. 向 DataNode 返回指令（如：删除某些 block、副本迁移等）

如果 NameNode 超过 dfs.heartbeat.timeout（默认 10.5 秒 = 3.0 × 3 × 1.5）没有收到心跳，就认为该 DataNode 已失联，触发副本重分配。

2.8. ✅ NameNode心跳总结

2.9. NameNode心跳采用sleep 是不是效率不高？ 为啥cronjob 来实现？

下面我从底层线程调度原理、性能考虑以及为什么 HDFS 不用 cronjob 来逐步回答你的问题。

2.9.1. ✅ Thread.sleep() 是否效率不高？

2.9.1.1. ❓ 它的本质是什么？

Thread.sleep(ms) 的本质是：

• 当前线程主动让出 CPU，进入 TIMED_WAITING 状态。
• 在指定时间后由系统调度器唤醒，继续运行。

2.9.1.2. ✅ 优点：

• 低 CPU 占用：线程在 sleep 时不消耗 CPU。
• 实现简单：只需 sleep 循环即可实现定时行为。

2.9.1.3. ⚠️ 缺点：

• 不是精准的定时器（受限于系统调度粒度）；
• 无法处理复杂的定时调度（如：每小时整点、每日0点等）；
• 对于大量线程 sleep-wake，系统调度器压力会增加。

但在 HDFS 的这种周期性任务中，这种方式其实是足够轻量且高效的。

2.9.2. ✅ 为什么不用 cron job？

你指的是系统级的 cron（Linux 中的定时任务）吗？HDFS 不用 cron 的原因如下：

2.9.2.1. 🚫 cron job 是系统级进程，不能高频运行

• cron 最小精度是 分钟级，不适合每 3 秒运行一次；
• 调用外部命令的方式也会引入启动开销。

2.9.2.2. 🚫 Hadoop 是多线程服务进程，不是脚本系统

• DataNode 是一个长时间运行的 Java 后台服务；
• 使用 cron 等外部定时器调度 Java 内部逻辑是不合适的。

2.9.3. ✅ 为什么不用Java的ScheduledExecutorService？

其实 Hadoop 也有一些地方使用了类似 ScheduledExecutorService，但在 DataNode 这种核心、频繁的心跳逻辑中：

• 为了最大控制权和性能，Hadoop 更倾向使用自己控制的循环 + sleep；
• 它也能更方便地处理异常、网络延迟、自适应重试等高级逻辑。

2.9.4. ✅ NameNode心跳sleep/cron总结

在 HDFS 中使用 Thread.sleep() 控制心跳周期，是一种 高效、可控、适合服务进程的方式。相比之下，cron job 不适合这种场景。

3. HDFS+SpringBoot实战示例

下面是一个完整的 Spring Boot + HDFS 实战示例，涵盖以下核心操作：

功能点包括：

• 上传文件到 HDFS
• 从 HDFS 下载文件
• 删除 HDFS 文件
• 列出目录下文件

3.1. 🧱 引入依赖（pom.xml）

<dependencies><!-- Spring Boot 核心依赖 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><!-- Hadoop HDFS 依赖 --><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>3.3.6</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-hdfs</artifactId><version>3.3.6</version></dependency><!-- Spring Web --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency>
</dependencies>

🔧 Hadoop 版本请与你的 HDFS 集群保持一致。

3.2. ⚙️ 配置文件 application.yml

hdfs:uri: hdfs://localhost:9000  # 替换为你集群的地址

3.3. 🧩 HDFS 配置类

@Configuration
public class HdfsConfig {@Value("${hdfs.uri}")private String hdfsUri;@Beanpublic FileSystem fileSystem() throws IOException {Configuration config = new Configuration();config.set("fs.defaultFS", hdfsUri);// 如果你使用的是 HA 或 Kerberos，需额外配置return FileSystem.get(URI.create(hdfsUri), config);}
}

3.4. 💼 四、HDFS 操作服务类

@Service
public class HdfsService {@Autowiredprivate FileSystem fileSystem;public void uploadFile(MultipartFile file, String hdfsPath) throws IOException {try (FSDataOutputStream output = fileSystem.create(new Path(hdfsPath));InputStream input = file.getInputStream()) {IOUtils.copyBytes(input, output, 4096, true);}}public byte[] downloadFile(String hdfsPath) throws IOException {try (FSDataInputStream input = fileSystem.open(new Path(hdfsPath));ByteArrayOutputStream output = new ByteArrayOutputStream()) {IOUtils.copyBytes(input, output, 4096, false);return output.toByteArray();}}public boolean deleteFile(String hdfsPath) throws IOException {return fileSystem.delete(new Path(hdfsPath), true);}public List<String> listFiles(String hdfsDir) throws IOException {FileStatus[] statuses = fileSystem.listStatus(new Path(hdfsDir));List<String> files = new ArrayList<>();for (FileStatus status : statuses) {files.add(status.getPath().toString());}return files;}
}

3.5. 📦 控制器层HdfsController

@RestController
@RequestMapping("/hdfs")
public class HdfsController {@Autowiredprivate HdfsService hdfsService;@PostMapping("/upload")public ResponseEntity<String> upload(@RequestParam MultipartFile file,@RequestParam String path) {try {hdfsService.uploadFile(file, path);return ResponseEntity.ok("上传成功: " + path);} catch (IOException e) {return ResponseEntity.status(500).body("上传失败: " + e.getMessage());}}@GetMapping("/download")public ResponseEntity<byte[]> download(@RequestParam String path) {try {byte[] content = hdfsService.downloadFile(path);return ResponseEntity.ok().header(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=" + Paths.get(path).getFileName()).body(content);} catch (IOException e) {return ResponseEntity.status(500).body(null);}}@DeleteMapping("/delete")public ResponseEntity<String> delete(@RequestParam String path) {try {boolean deleted = hdfsService.deleteFile(path);return ResponseEntity.ok(deleted ? "删除成功" : "删除失败");} catch (IOException e) {return ResponseEntity.status(500).body("删除失败: " + e.getMessage());}}@GetMapping("/list")public ResponseEntity<List<String>> list(@RequestParam String dir) {try {return ResponseEntity.ok(hdfsService.listFiles(dir));} catch (IOException e) {return ResponseEntity.status(500).body(Collections.emptyList());}}
}

4. HDFS优化设计

博文参考

《Hadoop原理》