HDFS 05 - HDFS 常用的 Java API 操作

0 - 配置 Hadoop 环境(Windows系统)

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

在 Windows 系统直接运行 Hadoop 相关代码,会提示缺少 winutils.exehadoop.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)

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

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

上一篇:使用java API操作hdfs--拷贝部分文件到本地


下一篇:使用Java API操作HDFS文件系统