【大数据存储与管理】分布式文件系统HDFS:06 HDFS的数据读写过程

【大数据存储与管理】分布式文件系统HDFS:06 HDFS的数据读写过程

【作者主页】Francek Chen
【专栏介绍】 ⌈ ⌈ ⌈大数据技术原理与应用 ⌋ ⌋ ⌋专栏系统介绍大数据的相关知识,分为大数据基础篇、大数据存储与管理篇、大数据处理与分析篇、大数据应用篇。内容包含大数据概述、大数据处理架构Hadoop、分布式文件系统HDFS、分布式数据库HBase、NoSQL数据库、云数据库、MapReduce、Hadoop再探讨、数据仓库Hive、Spark、流计算、Flink、图计算、数据可视化,以及大数据在互联网领域、生物医学领域的应用和大数据的其他应用。
【GitCode】专栏资源保存在我的GitCode仓库:https://gitcode.com/Morse_Chen/BigData_principle_application

文章目录


在介绍 HDFS 的数据读写过程之前,需要简单介绍一下相关的类。FileSystem 是一个通用文件系统的抽象基类,可以被分布式文件系统继承,所有可能使用 Hadoop 文件系统的代码都要使用到这个类。Hadoop 为 FileSystem 这个抽象类提供了多种具体的实现,DistributedFileSystem 就是 FileSystem 在 HDFS 中的实现。FileSystem 的 open()方法返回的是一个输入流 FSDataInputStream 对象,在 HDFS 中具体的输入流就是 DFSInputStream;FileSystem 中的 create()方法返回的是一个输出流 FSDataOutputStream 对象,在 HDFS 中具体的输出流就是 DFSOutputStream。

一、读数据的过程

importjava.io.BufferedReader;importjava.io.InputStreamReader;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.FSDataInputStream;importorg.apache.hadoop.fs.FileSystem;importorg.apache.hadoop.fs.Path;publicclassChapter3{publicstaticvoidmain(String[] args){try{Configuration conf =newConfiguration(); conf.set("fs.defaultFS","hdfs://localhost:9000"); conf.set("fs.hdfs.impl","org.apache.hadoop.hdfs.DistributedFileSystem");FileSystem fs =FileSystem.get(conf);Path file =newPath("test");FSDataInputStream getIt = fs.open(file);BufferedReader d =newBufferedReader(newInputStreamReader(getIt));String content = d.readLine();// 读取文件一行System.out.println(content); d.close();// 关闭文件 fs.close();// 关闭hdfs}catch(Exception e){ e.printStackTrace();}}}
Configuration conf =newConfiguration(); conf.set("fs.defaultFS","hdfs://localhost:9000"); conf.set("fs.hdfs.impl","org.apache.hadoop.hdfs.DistributedFileSystem");FileSystem fs =FileSystem.get(conf);FSDataInputStream in = fs.open(newPath(uri));FSDataOutputStream out = fs.create(newPath(uri));

客户端连续调用 open()、read()、close()读取数据时,HDFS 内部的执行过程如下图1。

在这里插入图片描述

图1 HDFS读数据的过程

  1. 客户端通过 FileSystem.open() 打开文件,相应地,在 HDFS 中 DistributedFileSystem 具体实现了 FileSystem。因此,调用 open() 方法后,DistributedFileSystem 会创建输入流 FSData InputStream,对于 HDFS 而言,具体的输入流就是 DFSInputStream。
  2. 在 DFSInputStream 的构造函数中,输入流通过 ClientProtocal.getBlockLocations() 远程调用名称节点,获得文件开始部分数据块的保存位置。对于该数据块,名称节点返回保存该数据块的所有数据节点的地址,同时根据距离客户端的远近对数据节点进行排序;然后,DistributedFileSystem 会利用 DFSInputStream 来实例化 FSDataInputStream,并返回给客户端,同时返回数据块的数据节点地址。
  3. 获得输入流 FSDataInputStream 后,客户端调用 read() 方法开始读取数据。输入流根据前面的排序结果,选择距离客户端最近的数据节点建立连接并读取数据。
  4. 数据从该数据节点读到客户端;当该数据块读取完毕时,FSDataInputStream 关闭和该数据节点的连接。
  5. 输入流通过 getBlockLocations() 方法查找下一个数据块(如果客户端缓存中已经包含了该数据块的位置信息,就不需要调用该方法)。
  6. 找到该数据块的最佳数据节点,读取数据。
  7. 当客户端读取完数据的时候,调用 FSDataInputStream 的 close() 方法,关闭输入流。需要注意的是,在读取数据的过程中,如果客户端与数据节点通信时出现错误,就会尝试连接包含此数据块的下一个数据节点。

二、写数据的过程

importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.FSDataOutputStream;importorg.apache.hadoop.fs.FileSystem;importorg.apache.hadoop.fs.Path;publicclassChapter3{publicstaticvoidmain(String[] args){try{Configuration conf =newConfiguration(); conf.set("fs.defaultFS","hdfs://localhost:9000"); conf.set("fs.hdfs.impl","org.apache.hadoop.hdfs.DistributedFileSystem");FileSystem fs =FileSystem.get(conf);byte[] buff ="Hello world".getBytes();// 要写入的内容String filename ="test";// 要写入的文件名FSDataOutputStream os = fs.create(newPath(filename)); os.write(buff,0, buff.length);System.out.println("Create:"+ filename); os.close(); fs.close();}catch(Exception e){ e.printStackTrace();}}}

客户端向 HDFS 写数据是一个复杂的过程,这里介绍一下在不发生任何异常的情况下,客户端连续调用 create()、write() 和 close() 时,HDFS 内部的执行过程见图2。

在这里插入图片描述

图2 HDFS写数据的过程

  1. 客户端通过 FileSystem.create() 创建文件,相应地,在 HDFS 中 Distributed FileSystem 具体实现了 FileSystem。因此,调用 create() 方法后,DistributedFileSystem 会创建输出流 FSDataOutputStream,对于 HDFS 而言,具体的输出流就是 DFSOutputStream。
  2. 然后,DistributedFileSystem 通过 RPC 远程调用名称节点,在文件系统的命名空间中创建一个新的文件。名称节点会执行一些检查,比如文件是否已经存在,客户端是否有权限创建文件等。检查通过之后,名称节点会构造一个新文件,并添加文件信息。远程方法调用结束后,DistributedFileSystem 会利用 DFSOutputStream 来实例化 FSDataOutputStream,并返回给客户端,客户端使用这个输出流写入数据。
  3. 获得输出流 FSDataOutputStream 以后,客户端调用输出流的 write() 方法向 HDFS 中对应的文件写入数据。
  4. 客户端向输出流 FSDataOutputStream 中写入的数据会首先被分成一个个的分包,这些分包被放入 DFSOutputStream 对象的内部队列。输出流 FSDataOutputStream 会向名称节点申请保存文件和副本数据块的若干个数据节点,这些数据节点形成一个数据流管道。队列中的分包最后被打包成数据包,发往数据流管道中的第 1 个数据节点,第 1 个数据节点将数据包发送给第 2 个数据节点,第 2 个数据节点将数据包发送给第 3 个数据节点,这样,数据包会流经管道上的各个数据节点。
  5. 因为各个数据节点位于不同的机器上,数据需要通过网络发送。因此,为了保证所有数据节点的数据都是准确的,接收到数据的数据节点要向发送者发送“确认包”(ACK Packet)。确认包沿着数据流管道逆流而上,从数据流管道依次经过各个数据节点并最终发往客户端,当客户端收到应答时,它将对应的分包从内部队列移除。不断执行 3~5 步,直到数据全部写完。
  6. 客户端调用 close() 方法关闭输出流,此时开始,客户端不会再向输出流中写入数据,所以,当 DFSOutputStream 对象内部队列中的分包都收到应答以后,就可以使用 ClientProtocol.complete() 方法通知名称节点关闭文件,完成一次正常的写文件过程。

小结

HDFS 读写数据时,读数据通过 FileSystem.open() 创建DFSInputStream,获取数据块位置,选择最近数据节点读取,读完关闭连接并查找下一数据块;写数据则通过 FileSystem.create() 创建 DFSOutputStream,远程调用名称节点创建文件,写入数据时分包放入队列,形成数据流管道传输,数据节点发送确认包,全部写完客户端调用 close() 关闭输出流,通知名称节点关闭文件,从而完成 HDFS 数据正常读写过程。

欢迎 点赞👍 | 收藏⭐ | 评论✍ | 关注🤗

Read more

K8S 之 Taints(污点)与 Tolerations(容忍)

K8S 之 Taints(污点)与 Tolerations(容忍)

👨‍🎓博主简介 🏅ZEEKLOG博客专家 🏅云计算领域优质创作者 🏅华为云开发者社区专家博主 🏅阿里云开发者社区专家博主 💊交流社区:运维交流社区 欢迎大家的加入! 🐋 希望大家多多支持,我们一起进步!😄 🎉如果文章对你有帮助的话,欢迎 点赞 👍🏻 评论 💬 收藏 ⭐️ 加关注+💗 文章目录 * 一、核心概念 * 1.1 什么是 Taints(污点) * 1.2 什么是 Tolerations(容忍) * 1.3 两者的关系 * 二、污点详解 * 2.1 污点语法格式 * 2.2 三种 Effect 效果 * 2.3 常用命令 * 2.4 生产场景示例(自定义标识) * 三、

By Ne0inhk
[特殊字符]颠覆MCP!Open WebUI新技术mcpo横空出世!支持ollama!轻松支持各种MCP Server!Cline+Claude3.7轻松开发论文检索MCP Server!

[特殊字符]颠覆MCP!Open WebUI新技术mcpo横空出世!支持ollama!轻松支持各种MCP Server!Cline+Claude3.7轻松开发论文检索MCP Server!

🔥🔥🔥本篇笔记所对应的视频:🚀颠覆MCP!Open WebUI新技术mcpo横空出世!支持ollama!轻松支持各种MCP Server!Cline+Claude3.7轻松开发MCP服务_哔哩哔哩_bilibili Open WebUI 的 MCPo 项目:将 MCP 工具无缝集成到 OpenAPI 的创新解决方案 随着人工智能工具和模型的快速发展,如何高效、安全地将这些工具集成到标准化的 API 接口中成为了开发者面临的重要挑战。Open WebUI 的 MCPo 项目(Model Context Protocol-to-OpenAPI Proxy Server)正是为了解决这一问题而设计的。本文将带您深入了解 MCPo 的功能、优势及其对开发者生态的影响。 什么是 MCPo? MCPo 是一个简单、可靠的代理服务器,能够将任何基于 MCP 协议的工具转换为兼容

By Ne0inhk
Qwen3+Qwen Agent 智能体开发实战,打开大模型MCP工具新方式!(一)

Qwen3+Qwen Agent 智能体开发实战,打开大模型MCP工具新方式!(一)

系列文章目录 一、Qwen3+Qwen Agent 智能体开发实战,打开大模型MCP工具新方式!(一) 二、Qwen3+Qwen Agent +MCP智能体开发实战(二)—10分钟打造"MiniManus" 前言 要说最近人工智能界最火热的开源大模型,必定是阿里发布不久的Qwen3系列模型。Qwen3模型凭借赶超DeepSeek-V3/R1的优异性能,创新的混合推理模式,以及极强的MCP能力迅速成为AI Agent开发的主流基座模型。大家可参考我的文章一文解析Qwen3大模型详细了解Qwen3模型的核心能力。有读者私信我: “Qwen3官网特地强调增强了Agent和代码能力,同时加强了对MCP的支持,那么我该如何利用Qwen3快速开发MCP应用呢?” 这就就需要使用我们今天的主角——Qwen官方推荐的开发工具Qwen-Agent ,本期分享我们就一起学习快速使用Qwen3+QwenAgent 接入MCP服务端,快速开发AI Agent应用! 一、注册 Qwen3 API-Key 本次分享通过阿里云百炼大模型服务平台API Key请求方式调用Qwen3大模型,获取服务平台

By Ne0inhk