【Hadoop_04】HDFS的API操作与读写流程

1、HDFS的API操作

1.1 客户端环境准备

• 首先要配置环境变量

• 其次在IDEA中创建一个Maven工程HdfsClientDemo，并导入相应的依赖坐标+日志添加

<properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target></properties><dependencies><dependency><groupId>org.hamcrest</groupId><artifactId>hamcrest-core</artifactId><version>1.3</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>3.1.3</version></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.12</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>1.7.30</version></dependency></dependencies></project>

1.2 API创建文件夹

packagecom.wenxin.hdfs;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.FileSystem;importorg.apache.hadoop.fs.Path;importorg.junit.After;importorg.junit.Before;importorg.junit.Test;importjava.io.IOException;importjava.net.URI;importjava.net.URISyntaxException;/**
 * @author Susie-Wen
 * @version 1.0
 * @description:客户端代码常用套路
 * 1、获取一个客户端对象
 * 2、执行相关的操作命令
 * 3、关闭资源
 * HDFS zookeeper
 * @date 2023/12/11 12:27
 */publicclassHdfsClient{privateFileSystem fs;@Beforepublicvoidinit()throwsURISyntaxException,IOException,InterruptedException{// 连接的集群地址URI uri =newURI("hdfs://hadoop102:8020");// 用户String user ="root";// 创建一个配置文件Configuration configuration =newConfiguration();// 1、获取到了客户端对象
        fs =FileSystem.get(uri, configuration, user);}@Afterpublicvoidclose()throwsIOException{// 3、关闭资源
        fs.close();}@TestpublicvoidtestMkdir()throwsIOException{// 2、创建一个文件夹
        fs.mkdirs(newPath("/xiyou/huaguoshan"));}}

上面这段代码把连接和关闭资源都进行了封装，更加方便。

• @Before 注解标识的方法 init() 是一个在测试方法执行之前会被调用的初始化方法。
• @After 注解标识的方法 close() 是一个在测试方法执行之后会被调用的清理方法。

如下所示，确实创建了文件夹

1.3 API上传

接下来进行API上传操作：使用客户端远程访问HDFS，之后上传文件。

// 上传：客户端远程访问HDFS,之后上传文件@TestpublicvoidtestPut()throwsIOException{
        fs.copyFromLocalFile(false,false,newPath("E:\\VMWare\\Centos\\sunwukong.txt"),newPath("hdfs://hadoop102/xiyou/huaguoshan"));}

1.4 API参数的优先级

HDFS文件上传（测试参数优先级）

1）编写源代码

@TestpublicvoidtestCopyFromLocalFile()throwsIOException,InterruptedException,URISyntaxException{// 1 获取文件系统Configuration configuration =newConfiguration();
    configuration.set("dfs.replication","2");FileSystem fs =FileSystem.get(newURI("hdfs://hadoop102:8020"), configuration,"atguigu");// 2 上传文件
    fs.copyFromLocalFile(newPath("d:/sunwukong.txt"),newPath("/xiyou/huaguoshan"));// 3 关闭资源
    fs.close();
｝

2）将hdfs-site.xml拷贝到项目的resources资源目录下

<?xml version="1.0" encoding="UTF-8"?><?xml-stylesheet type="text/xsl" href="configuration.xsl"?><configuration><property><name>dfs.replication</name><value>1</value></property></configuration>

参数优先级排序：（1）客户端代码中设置的值 >（2）ClassPath下的用户自定义配置文件 >（3）然后是服务器的自定义配置（xxx-site.xml） >（4）服务器的默认配置（xxx-default.xml）

1.5 API文件夹下载

下载相当于从HDFS将文件下载到windows本地：

//下载：将文件从HDFS下载到windows当中publicvoidtestGet()throwsIOException{
        fs.copyToLocalFile(false,newPath("hdfs://hadoop102/xiyou/huaguoshan"),newPath("E:\\VMWare\\"),false);}

• 如果参数四设置为true的话，就不会进行crc校验

1.6 API文件删除

//删除@TestpublicvoidtestRm()throwsIOException{
        fs.delete(newPath("/xiyou/huaguoshan/sunwukong.txt"),false);}

• 除了删除文件之外，我们还可以删除空目录以及非空目录
• 多个文件如果是非递归删除的话，会报错

//删除@TestpublicvoidtestRm()throwsIOException{//1.删除文件
        fs.delete(newPath("/xiyou/huaguoshan/sunwukong.txt"),false);//2.删除空目录
        fs.delete(newPath("/xiyou"),false);//3.删除非空目录
        fs.delete(newPath("/xiyou/huaguoshan/"),false);}

1.7 API文件更名和移动

• 包括文件名称的修改，文件的移动和更名以及目录的更名

//文件的更名和移动@TestpublicvoidtestMove()throwsIOException{//1.文件名称的修改
        fs.rename(newPath("/input/word.txt"),newPath("/input/ss.txt"));//2.文件的移动和更名:从input目录移动到根目录下并修改姓名
        fs.rename(newPath("/input/ss.txt"),newPath("/wenxin.txt"));//3.目录的更名
        fs.rename(newPath("/input"),newPath("/output"));}

1.8 API文件详情和查看

查看文件名称、权限、长度、块信息

//获取文件详情信息@TestpublicvoidfileDetail()throwsIOException{//1.获取所有文件信息RemoteIterator<LocatedFileStatus> listFiles=fs.listFiles(newPath("/"),true);//2.遍历文件while(listFiles.hasNext()){LocatedFileStatus fileStatus=listFiles.next();System.out.println("====="+fileStatus.getPath()+"=====");System.out.println(fileStatus.getPermission());System.out.println(fileStatus.getOwner());System.out.println(fileStatus.getPath());System.out.println(fileStatus.getModificationTime());System.out.println(fileStatus.getReplication());System.out.println(fileStatus.getBlockLocations());System.out.println(fileStatus.getPath().getName());}}

//获取文件详情信息@TestpublicvoidfileDetail()throwsIOException{//1.获取所有文件信息RemoteIterator<LocatedFileStatus> listFiles=fs.listFiles(newPath("/"),true);//2.遍历文件while(listFiles.hasNext()){LocatedFileStatus fileStatus=listFiles.next();System.out.println("====="+fileStatus.getPath()+"=====");System.out.println(fileStatus.getPermission());System.out.println(fileStatus.getOwner());System.out.println(fileStatus.getPath());System.out.println(fileStatus.getModificationTime());System.out.println(fileStatus.getReplication());System.out.println(fileStatus.getBlockLocations());System.out.println(fileStatus.getPath().getName());//获取块信息BlockLocation[] blockLocations = fileStatus.getBlockLocations();System.out.println(Arrays.toString(blockLocations));}}

1.9 API文件和文件夹判断

判断根目录那个是文件，那个是文件夹

//判断文件与文件夹@TestpublicvoidtestFile()throwsIOException{FileStatus[] listStatus = fs.listStatus(newPath("/"));for(FileStatus status:listStatus){if(status.isFile()){System.out.println("文件："+status.getPath().getName());}else{System.out.println("目录："+status.getPath().getName());}}}

2、HDFS的读写流程（面试重点）

2.1 HDFS写数据流程

（1）客户端通过Distributed FileSystem模块向NameNode请求上传文件，NameNode检查目标文件是否已存在，父目录是否存在。

（2）NameNode返回是否可以上传。

（3）客户端请求第一个 Block上传到哪几个DataNode服务器上。

（4）NameNode返回3个DataNode节点，分别为dn1、dn2、dn3。

（5）客户端通过FSDataOutputStream模块请求dn1上传数据，dn1收到请求会继续调用dn2，然后dn2调用dn3，将这个通信管道建立完成。

（6）dn1、dn2、dn3逐级应答客户端。

（7）客户端开始往dn1上传第一个Block（先从磁盘读取数据放到一个本地内存缓存），以Packet为单位，dn1收到一个Packet就会传给dn2，dn2传给dn3；dn1每传一个packet会放入一个应答队列等待应答。

（8）当一个Block传输完成之后，客户端再次请求NameNode上传第二个Block的服务器。（重复执行3-7步）。

2.2 网络拓扑-节点距离计算

在HDFS写数据的过程中，NameNode会选择距离待上传数据最近距离的DataNode接收数据。那么这个最近距离怎么计算呢？

节点距离：两个节点到达最近的共同祖先的距离总和。

例如，假设有数据中心d1机架r1中的节点n1。该节点可以表示为/d1/r1/n1。利用这种标记，这里给出四种距离描述。

算一算每两个节点之间的距离：

2.3 机架感知（副本存储节点选择）

官方说明：

对于常见情况，当副本为3时，HDFS的放置策略是，如果编写器在datanode上，则将一个副本放在本地计算机上，否则放在随机datanode上，另一个副本放在不同(远程)机架中的节点上，最后一个放在同一远程机架中的不同节点上。此策略减少了机架间的写入流量，从而总体上提高了写入性能。机架故障的几率远小于节点故障的几率；该策略不影响数据可靠性和可用性保证。但是，它确实减少了读取数据时使用的聚合网络带宽，因为一个数据块只放在两个不同的机架中，而不是三个。使用此策略，文件的副本不会均匀分布在机架上。三分之一的副本位于一个节点上，三分之二的副本位于一个机架上，另外三分之一的副本均匀分布在其余机架上。该策略提高了写入性能，而不影响数据可靠性或读取性能。

• 第一个副本考虑的是节点距离最近，上传速度最快。
• 第二个节点保证数据的可靠性。
• 第三个节点在保证数据可靠性的前提下兼顾效率。

查看源码：

Crtl + n 查找BlockPlacementPolicyDefault，在该类中查找chooseTargetInOrder方法。

2.4 读数据流程

（1）客户端通过DistributedFileSystem向NameNode请求下载文件，NameNode通过查询元数据，找到文件块所在的DataNode地址。【DistributedFileSystem是分布式文件系统对象】

（2）挑选一台DataNode（就近原则，然后随机）服务器，请求读取数据。【除了考虑节点最近之外，还会考虑当前节点的负载能力】

（3）DataNode开始传输数据给客户端（从磁盘里面读取数据输入流，以Packet为单位来做校验）。

（4）客户端以Packet为单位接收，先在本地缓存，然后写入目标文件。

• 这里读取数据采用的是串行读取，而不是并行读取。