HDFS 05 - HDFS 常用的 Java API 操作

2023-08-15 09:07:52

0 - 配置 Hadoop 环境（Windows系统）
1 - 导入 Maven 依赖
2 - 常用类介绍
3 - 常见 API 操作
4 - HDFS 的访问权限控制
版权声明

0 - 配置 Hadoop 环境（Windows系统）

下述步骤适用于 Windows 系统，其他系统可忽略。

在 Windows 系统直接运行 Hadoop 相关代码，会提示缺少 winutils.exe 和 hadoop.dll 文件：

java.io.IOException: Could not locate executable null\bin\winutils.exe in the Hadoop binaries.

WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable

原因：通过代码访问 Hadoop 集群，本地开发环境相当于 Hadoop 客户端，需要有 Hadoop 相关软件才可正常运行。

配置步骤：

1）到 https://github.com/cdarlint/winutils 下载与集群版本相匹配的文件夹，然后将此文件夹拷贝到没有中文和空格的路径下，比如 D:\software\hadoop-3.2.1；

2）在 Windows 的环境变量中添加 HADOOP_HOME，值为上面的路径，并将 %HADOOP_HOME%\bin 添加到 path 中；

3）把上述文件夹 bin目录下的 hadoop.dll 文件拷贝到系统盘 C:\Windows\System32 目录；

4）重启 Windows 电脑。

1 - 导入 Maven 依赖

鉴于篇幅有限，相关 Maven 依赖请参见：《https://github.com/healchow/bigdata-study/blob/main/pom.xml》

2 - 常用类介绍

通过 Java API 操作 HDFS，主要涉及以下 class：

1）Configuration

主要用来封装客户端 / 服务端的配置。

2）FileSystem

这个类的对象是一个文件系统对象，可以用该对象的一些方法来对文件进行操作。

可通过静态方法获得该对象：

// 通过 conf 中的 “fs.defaultFS” 参数的值来确定文件系统的具体类型

FileSystem fs = FileSystem.get(conf);

如果代码中没有指定 fs.defaultFS，并且工程的 ClassPath 下也没有相应的配置，此参数的默认值就由 Hadoop Jar 包中的 core-default.xml 文件来确定：

默认值是 file:/// ，获取的不是 DistributedFileSystem 的实例，而是一个本地文件系统的客户端对象。

3 - 常见 API 操作

3.1 获取文件系统（重要）

方式1：FileSystem.get(conf)

/**

 * 获取 FileSystem - FileSystem.get()

 */

@Test

public void testGetFileSystem1() throws IOException {

    // 创建 Configuration 对象

    Configuration conf = new Configuration();

    // 指定文件系统类型

    conf.set("fs.defaultFS", "hdfs://hadoop:9000");

    // 获取指定的文件系统

    FileSystem fileSystem = FileSystem.get(conf);

    // FileSystem fileSystem = FileSystem.get(new URI("hdfs://hadoop:9000"), new Configuration());

    // 结果：DFS[DFSClient[clientName=DFSClient_NONMAPREDUCE_1219793882_1, ugi=healchow (auth:SIMPLE)]]

    System.out.println(fileSystem);

    // 关闭文件系统

    fileSystem.close();

}

方式2：FileSystem.newInstance(conf)

/**

 * 获取 FileSystem - FileSystem.newInstance()

 */

@Test

public void testGetFileSystem2() throws IOException {

    // 创建 Configuration 对象

    Configuration conf = new Configuration();

    // 指定文件系统类型

    conf.set("fs.defaultFS", "hdfs://hadoop:9000");

    // 获取指定的文件系统

    FileSystem fileSystem = FileSystem.newInstance(conf);

    // FileSystem fileSystem = FileSystem.newInstance(new URI("hdfs://hadoop:9000"), new Configuration());

    System.out.println(fileSystem);

    fileSystem.close();

}

3.2 创建目录、写入文件

/**

 * 通过 HDFS URL 创建目录、写入文件

 */

@Test

public void testPutFile() throws IOException, URISyntaxException {

    // 创建测试目录（可创建多级目录）

    FileSystem fileSystem = FileSystem.newInstance(new URI("hdfs://hadoop:9000"), new Configuration());

    boolean result = fileSystem.mkdirs(new Path("/test/input"));

    System.out.println("mkdir result: " + result);

    // 创建文件，若存在则覆盖，返回的是写入文件的输出流

    FSDataOutputStream outputStream = fileSystem.create(new Path("/test/input/hello.txt"), true);

    String content = "hello,hadoop\nhello,hdfs";

    outputStream.write(content.getBytes(StandardCharsets.UTF_8));

    // 关闭流（不抛出异常）

    IOUtils.closeQuietly(outputStream);

}

3.3 上传文件

/**

 * 向 HDFS 上传文件 - copyFromLocalFile()

 */

@Test

public void testUploadFile() throws URISyntaxException, IOException {

    // 获取 FileSystem

    FileSystem fileSystem = FileSystem.get(new URI("hdfs://hadoop:9000"), new Configuration());

    // 从本地上传文件，两个参数都要指定到具体的文件

    fileSystem.copyFromLocalFile(new Path("/Users/healchow/bigdata/core-site.xml"),

            new Path("/test/upload/core-site.xml"));

    // 关闭FileSystem

    fileSystem.close();

}

3.4 下载文件

HDFS URL 打开 InputStream 的方式：

/**

 * 通过 HDFS URL 获取文件并下载 - IOUtils.copy() 方法

 */

@Test

public void testDownFileByUrl() throws IOException {

    // 注册 HDFS URL

    URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());

    // 获取 HDFS 文件的输入流

    InputStream inputStream = new URL("hdfs://hadoop:9000/test/input/hello.txt").openStream();

    // 获取本地文件的输出流（绝对路径，文件夹必须存在）

    FileOutputStream outputStream = new FileOutputStream("/Users/healchow/bigdata/test/hello.txt");

    // 拷贝文件

    IOUtils.copy(inputStream, outputStream);

    // 关闭流（不抛出异常）

    IOUtils.closeQuietly(inputStream);

    IOUtils.closeQuietly(outputStream);

}

FileSystem 打开 InputStream 的方式：

/**

 * 通过 FileSystem 获取文件并下载 - IOUtils.copy() 方法

 */

@Test

public void testDownloadFile() throws URISyntaxException, IOException {

    // 获取 FileSystem

    FileSystem fileSystem = FileSystem.get(new URI("hdfs://hadoop:9000"), new Configuration());

    // 获取 HDFS 文件的输入流

    FSDataInputStream inputStream = fileSystem.open(new Path("/test/input/hello.txt"));

    // 获取本地文件的输出流

    FileOutputStream outputStream = new FileOutputStream("/Users/healchow/bigdata/test/hello1.txt");

    // 拷贝文件

    IOUtils.copy(inputStream, outputStream);

    // 关闭流

    IOUtils.closeQuietly(inputStream);

    IOUtils.closeQuietly(outputStream);

    fileSystem.close();

}

FileSystem#copyToLocalFile() 的方式：

/**

 * 通过 FileSystem 获取文件并下载 - copyToLocalFile() 方法

 */

@Test

public void testDownloadFileByCopyTo() throws URISyntaxException, IOException, InterruptedException {

    // 获取 FileSystem

    FileSystem fileSystem = FileSystem.get(new URI("hdfs://hadoop:9000"), new Configuration(), "root");

    // copyToLocalFile 拷贝文件到本地，会下载 CRC 校验文件

    fileSystem.copyToLocalFile(new Path("/test/input/hello.txt"),

            new Path("/Users/healchow/bigdata/test/hello2.txt"));

    // 关闭 FileSystem

    fileSystem.close();

}

3.5 遍历 HDFS 的文件

/**

 * 遍历 HDFS 文件

 */

@Test

public void testListFiles() throws URISyntaxException, IOException {

    // 获取FileSystem实例

    FileSystem fileSystem = FileSystem.get(new URI("hdfs://hadoop:9000"), new Configuration());

    // 递归获取 /test 目录下所有的文件信息

    RemoteIterator<LocatedFileStatus> iterator = fileSystem.listFiles(new Path("/test"), true);

    // 遍历文件

    while (iterator.hasNext()) {

        LocatedFileStatus fileStatus = iterator.next();

        // 获取文件的绝对路径：hdfs://hadoop:9000/xxx

        System.out.println("filePath: " + fileStatus.getPath());

        // 文件的 block 信息

        BlockLocation[] blockLocations = fileStatus.getBlockLocations();

        for (BlockLocation blockLocation : blockLocations) {

            String[] hosts = blockLocation.getHosts();

            for (String host : hosts) {

                System.out.println("blockHost: " + host);

            }

        }

        System.out.println("blockSize: " + blockLocations.length);

    }

}

4 - HDFS 的访问权限控制

从上面的 API 练习，不难发现：只要得到了 HDFS 的 URL（即 fs.defaultFS）配置项，能访问到集群的任何人都能读写 HDFS 上的数据，这会导致数据的安全性完全无法得到保障。

为了解决这个问题，HDFS 有访问权限控制的方法，只有通过认证的用户，按照其所拥有的权限，读取或写入某些目录下的文件。

开启 HDFS 访问权限控制的方法如下：

1）停止 HDFS 集群：

cd ~/bigdata/hadoop-3.2.1

sbin/stop-dfs.sh

2）修改 ~/bigdata/hadoop-3.2.1/etc/hadoop/hdfs-site.xml 中的配置，添加如下内容：

<property>

    <name>dfs.permissions.enabled</name>

    <value>true</value>

</property>

4）重启 HDFS 集群：

cd ~/bigdata/hadoop-3.2.1

sbin/start-dfs.sh

5）上传测试文件到 HDFS 集群，这里将上传后的一个文件的权限修改为 600，即只能所有者读写：

cd ~/bigdata/hadoop-3.2.1/etc/hadoop

hdfs dfs -mkdir /test/config

hdfs dfs -put *.xml /test/config

hdfs dfs -chmod 600 /test/config/core-site.xml

6）通过代码下载文件：

/**

 * 通过下载文件，测试访问权限控制

 */

@Test

public void testAccessControl() throws Exception {

    // 开启权限控制后，当前用户（启动 NameNode 的用户）应当能成功访问

    // FileSystem fileSystem = FileSystem.get(new URI("hdfs://hadoop:9000"), new Configuration());

    // 伪造其他用户访问，应当访问失败

    FileSystem fileSystem = FileSystem.get(new URI("hdfs://hadoop:9000"), new Configuration(), "testuser");

    fileSystem.copyToLocalFile(new Path("/test/config/core-site.xml"),

            new Path("file:/Users/healchow/bigdata/core-site.xml"));

    fileSystem.close();

}

说明：本地测试失败。无论用哪个用户，访问都成功。

查了很多资料，没有说得通的。劳烦有了解的大佬，留言告知我呀

版权声明

作者：瘦风(https://healchow.com)

出处：博客园-瘦风的南墙(https://www.cnblogs.com/shoufeng)

感谢阅读，公众号「瘦风的南墙」，手机端阅读更佳，还有其他福利和心得输出，欢迎扫码关注

本文版权归博主所有，欢迎转载，但 [必须在页面明显位置标明原文链接]，否则博主保留追究相关人士法律责任的权利。

码农公寓