Java操作HDFS系统 API

开发环境简介和配置

本人的环境如下

操作系统：Windows 11 专业工作站版 64位
Maven版本：3.8.3
JDK版本：17.0.7
Hadoop版本：3.3.6

配置Windows Native Hadoop library

不配置的话，会报错：Unable to load native-hadoop library for your platform... using builtin-java classes where applicable

提示

下载地址很错乱，截止到 2024-4-5，本人只找到了3.3.6版本的windows系统的hadoop库

原作者不再更新windows库，转交给了 cdarlint大佬，仓库地址：https://github.com/cdarlint/winutils

但是这个人又没更新，我在仓库的 Pull requests 里找到了相关的 haijialiu大佬 的仓库地址：https://github.com/haijialiu/winutils

还不太清楚之后的更新是不是需要手动编译原代码

Windows的编译可以查看这个链接：https://www.jianshu.com/p/1b4cbabfd899

看着挺靠谱的，但是作者我很懒，没试过

下载下来之后，将对应的版本的文件夹（hadoop-版本号）找个地方放好，不要有中文路径，然后配置环境变量

HADOOP_HOME：hadoop库的根目录

PATH：%HADOOP_HOME%\bin

项目环境配置

Maven依赖地址：https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-client

Maven依赖的版本必须和hadoop版本一致，如下

<!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-client -->
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>3.3.6</version>
</dependency>

其他相关依赖

<!-- 单元测试 -->
<dependency>
    <groupId>junit</groupId>
    <artifactId>junit</artifactId>
    <version>4.13.2</version>
    <scope>test</scope>
</dependency>
<!-- 用来打印日志 -->
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-log4j12</artifactId>
    <version>1.7.36</version>
    <type>pom</type>
    <scope>test</scope>
</dependency>

然后在 src/main/resources 下创建配置文件 log4j.properties

log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/spring.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

代码通用配置

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;

public class HdfsClient {

    private FileSystem fs;

    @Before
    public void init() throws URISyntaxException, IOException, InterruptedException {
        // 这里的Client应该访问的是集群中NameNode的地址，因为NameNode才是集群中的“老板”
        // 直接访问testHadoop1，需要配置hosts文件
        URI uri = new URI("hdfs://testHadoop1:8020");
        // 创建一个配置文件类
        Configuration conf = new Configuration();
        // 指定用户
        String user = "root";
        // 获取客户端对象
        fs = FileSystem.get(uri, conf, user);
    }

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

}

写好这些代码，每次在执行测试用例的时候，就都会先init，再执行test，最后自动close

mkdir 创建文件夹

    @Test
    public void testMkdir() throws IOException {
        // 第一个参数：需要创建的文件夹地址
        fs.mkdirs(new Path("/aDir"));
    }

put copyFromLocal 上传文件

    @Test
    public void testPut() throws IOException {
        // 第一个参数：是否删除原数据。true的话，上传文件后删除本地文件，类似于moveFromLocal
        // 第二个参数：是否覆盖文件。false的话，如果上传位置存在同名文件，则会报错
        // 第三个参数：本地文件路径
        // 第四个参数：HDFS目标文件夹路径

        // 这里是上传到了 /aDir 目录下
        fs.copyFromLocalFile(false, false, new Path("D:\\tmp\\test.txt"), new Path("/aDir"));
    }

配置的优先级问题

刚才在上面上传了一个副本数量为 3 的 test.txt 文件，副本默认数量是在 hdfs-default.xml 中设置的。这个是hadoop默认的配置，想要更改这个配置文件只能重新编译源代码

如果我把 hadoop 的 hdfs-site.xml 配置文件修改一下，新的配置内容就会覆盖掉默认的配置内容（没在 hdfs-site.xml 文件里写的配置不覆盖）。也就是说：hdfs-site.xml 文件 优先级大于 hdfs-default.xml 文件。这个是已经尝试过的了

现在，我承接上个标题内的put代码，在resources文件夹下创建文件 hdfs-site.xml，尝试比较一下代码资源目录下的 hdfs-site.xml 配置文件和hadoop集群里的 hdfs-site.xml 配置文件的优先级哪个更高（是的，你没看错，他们两个是同名文件）

内容如下

<?xml version="1.0" encoding="UTF-8" ?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>

<configuration>
    <property>
        <!-- 配置副本数量 -->
        <name>dfs.replication</name>
        <value>2</value>
    </property>
</configuration>

提示

hadoop集群中的 hdfs-site.xml 文件里已经显式配置过副本数量了，所以不用担心是否覆盖的是 hdfs-default.xml 文件

删除 HDFS 中已经上传的 test.txt 文件，准备好代码和配置文件，重新上传

可以发现，新上传的文件的副本数量是2，也就是说明了：在代码资源目录下的配置文件 优先级大于 hadoop集群中同名的配置文件

再来进行个小尝试：init方法的代码中通过 Configuration 类创建了一个 名为 conf 的 配置文件对象

这个对象里面是可以写设置的，在创建对象后，添加如下代码

// 设置副本数量为1
conf.set("dfs.replication", "1");

再次删除 HDFS 中刚才上传的 test.txt 文件，准备好代码，重新上传

可以发现，上传上去的文件的副本数量是1，那么可以得出结论：代码中配置文件的对象 优先级大于 代码资源目录下的配置文件

总结一下

• 代码中配置文件的对象 优先级大于 代码资源目录下的配置文件
• 在代码资源目录下的配置文件 优先级大于 hadoop集群中同名的配置文件
• XXXX-site.xml 文件 优先级大于 XXXX-default.xml 文件

测试完成后，不要忘记将代码中的配置还原回去

get copyToLocal 下载文件

    @Test
    public void testGet() throws IOException {
        // 第一个参数：是否删除原数据。true的话，下载文件后删除HDFS中的文件
        // 第二个参数：HDFS文件路径
        // 第三个参数：本地文件夹目标路径
        /*
           第四个参数：是否开启本地文件校验，默认为false。
           false是开启了文件校验，为了确保传输过来的数据是没有问题的（万一传过来的数据丢了1字节），除了对应的文件外还会再传输一个crc文件
           这个参数基本不用
         */
        fs.copyToLocalFile(false, new Path("/aDir/test.txt"), new Path("D:\\tmp"), false);
    }

crc相关博客：

• https://blog.csdn.net/tianyeshiye/article/details/85293295
• https://zhuanlan.zhihu.com/p/346792759

rm 删除文件

    @Test
    public void testRm() throws IOException {
        // 第一个参数：需要删除的文件路径
        // 第二个参数：是否递归删除。删除的是文件夹的话，如果文件夹下有内容，可以为true
        fs.delete(new Path("/aDir"), true);
    }

mv 文件的更名和移动

    @Test
    public void testMv() throws IOException {
        // 第一个参数：原文件路径
        // 第二个参数：目标文件路径

        // 这个方法是hdfs系统的内部操作
        fs.rename(new Path("/aDir/test.txt"), new Path("/aDir/test1.txt"));
    }

获取文件详细信息

我想通过代码的方式，读取到web页面上展示出来的文件信息

    @Test
    public void testFileDetail() throws IOException {
        // 获取所有文件信息

        // 第一个参数，根据指定路径获取文件
        // 第二个参数，是否递归查找
        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);

        // 遍历文件
        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.getGroup());
            System.out.println(fileStatus.getLen());
            System.out.println(fileStatus.getModificationTime());
            System.out.println(fileStatus.getReplication());
            System.out.println(fileStatus.getBlockSize());
            System.out.println(fileStatus.getPath().getName());

            // 获取块信息
            BlockLocation[] blockLocations = fileStatus.getBlockLocations();
            /*
               当前输出的信息每5个为一组，后3个是因为副本数量为3，展示出数据在哪3个服务器上存储着
               前2个数据代表着从什么位置开始读取，读取多少字节 数据
             */
            System.out.println(Arrays.toString(blockLocations));
        }
    }

文件和文件夹判断

    @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() + " 是个文件夹");
            }
        }
    }