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

2023-12-13 05:43:37

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创建文件夹

package com.wenxin.hdfs;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

/**
 * @author Susie-Wen
 * @version 1.0
 * @description:客户端代码常用套路
 * 1、获取一个客户端对象
 * 2、执行相关的操作命令
 * 3、关闭资源
 * HDFS zookeeper
 * @date 2023/12/11 12:27
 */
public class HdfsClient {
    private FileSystem fs;

    @Before
    public void init() throws URISyntaxException, IOException, InterruptedException {
        // 连接的集群地址
        URI uri = new URI("hdfs://hadoop102:8020");
        // 用户
        String user = "root";
        // 创建一个配置文件
        Configuration configuration = new Configuration();
        // 1、获取到了客户端对象
        fs = FileSystem.get(uri, configuration, user);
    }

    @After
    public void close() throws IOException {
        // 3、关闭资源
        fs.close();
    }

    @Test
    public void testMkdir() throws IOException {
        // 2、创建一个文件夹
        fs.mkdirs(new Path("/xiyou/huaguoshan"));
    }
}

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

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

在这里插入图片描述

如下所示,确实创建了文件夹
在这里插入图片描述
在这里插入图片描述

1.3 API上传

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

在这里插入图片描述

    // 上传:客户端远程访问HDFS,之后上传文件
    @Test
    public void testPut() throws IOException {
        fs.copyFromLocalFile(false,false,
                new Path("E:\\VMWare\\Centos\\sunwukong.txt"),
                new Path("hdfs://hadoop102/xiyou/huaguoshan"));
    }

1.4 API参数的优先级

HDFS文件上传(测试参数优先级)

1)编写源代码

@Test
public void testCopyFromLocalFile() throws IOException, InterruptedException, URISyntaxException {

    // 1 获取文件系统
    Configuration configuration = new Configuration();
    configuration.set("dfs.replication", "2");
    FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:8020"), configuration, "atguigu");

    // 2 上传文件
    fs.copyFromLocalFile(new Path("d:/sunwukong.txt"), new Path("/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当中
    public void testGet() throws IOException {
        fs.copyToLocalFile(false,new Path("hdfs://hadoop102/xiyou/huaguoshan"),
        new Path("E:\\VMWare\\"),false);
    }
  • 如果参数四设置为true的话,就不会进行crc校验
    在这里插入图片描述

1.6 API文件删除

在这里插入图片描述

    //删除
    @Test
    public void testRm() throws IOException {
        fs.delete(new Path("/xiyou/huaguoshan/sunwukong.txt"),false);
    }
  • 除了删除文件之外,我们还可以删除空目录以及非空目录
  • 多个文件如果是非递归删除的话,会报错
    //删除
    @Test
    public void testRm() throws IOException {
        //1.删除文件
        fs.delete(new Path("/xiyou/huaguoshan/sunwukong.txt"),false);
        //2.删除空目录
        fs.delete(new Path("/xiyou"),false);
        //3.删除非空目录
        fs.delete(new Path("/xiyou/huaguoshan/"),false);
    }

1.7 API文件更名和移动

在这里插入图片描述

  • 包括文件名称的修改,文件的移动和更名以及目录的更名
    //文件的更名和移动
    @Test
    public void testMove() throws IOException {
        //1.文件名称的修改
        fs.rename(new Path("/input/word.txt"),new Path("/input/ss.txt"));
        //2.文件的移动和更名:从input目录移动到根目录下并修改姓名
        fs.rename(new Path("/input/ss.txt"),new Path("/wenxin.txt"));
        //3.目录的更名
        fs.rename(new Path("/input"),new Path("/output"));
    }

1.8 API文件详情和查看

在这里插入图片描述
在这里插入图片描述

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

    //获取文件详情信息
    @Test
    public void fileDetail() throws IOException {
        //1.获取所有文件信息
        RemoteIterator<LocatedFileStatus> listFiles=fs.listFiles(new Path("/"), 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());
        }
    }

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

    //获取文件详情信息
    @Test
    public void fileDetail() throws IOException {
        //1.获取所有文件信息
        RemoteIterator<LocatedFileStatus> listFiles=fs.listFiles(new Path("/"), 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文件和文件夹判断

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

    //判断文件与文件夹
    @Test
    public void testFile() throws IOException{
        FileStatus[] listStatus = fs.listStatus(new Path("/"));
        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为单位接收,先在本地缓存,然后写入目标文件。

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

文章来源:https://blog.csdn.net/wxfighting/article/details/134934135
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。