---

前言

本文将详细介绍如何使用Java编程语言与HDFS进行交互。我们将通过创建一个简单的Maven项目，逐步演示HDFS的常用Java API，包括创建目录、上传和下载文件、查看文件内容、删除文件等操作。通过这些示例，读者将能够掌握基本的HDFS操作，并为后续的大数据处理打下坚实的基础。

---

一、创建hdfs-demo项目

1. 在idea上创建maven项目

打开idea新建项目，如下图。

选择Java项目，输入项目名称，选择构建系统为Maven，选择JDK为1.8，然后点击创建。

2. 导入hadoop相关依赖

如下图所示，在pom.xml文件中添加Hadoop相关依赖。

---

二、常用 HDFS Java API

1. 简介

Hadoop分布式文件系统（HDFS）是Apache Hadoop的核心组件之一，设计用于存储大量的数据，并提供高吞吐量的数据访问。HDFS Java API为开发者提供了直接与HDFS交互的能力，允许执行诸如创建目录、上传和下载文件、读取和写入文件等操作。

2. 获取文件系统实例

编写一个getFs方法，用于返回一个文件系统实例。

public static FileSystem getFs() throws IOException, InterruptedException {// 1. 创建Hadoop配置对象，并设置配置信息Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://192.168.121.100:9000");conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");// 2. 获取文件系统实例，设置文件路径return FileSystem.get(URI.create(""), conf, "root");
}

3. 创建目录

可以使用mkdirs()方法来创建新的目录。

package org.example;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;import java.io.IOException;
import java.net.URI;public class Main {public static void main(String[] args) throws IOException, InterruptedException {FileSystem fs = getFs();// 创建目录/test/demoboolean mkdirs = fs.mkdirs(new Path("/test/demo"));if (mkdirs) {System.out.println("===============创建目录成功===============");} else {System.out.println("===============创建目录失败===============");}// 创建目录/test/demo2，并设置权限为777boolean mkdirs2 = fs.mkdirs(new Path("/test/demo2"), new FsPermission("777"));if (mkdirs2) {System.out.println("===============创建目录并设置权限成功===============");} else {System.out.println("===============创建目录并设置权限失败===============");}// 关闭文件系统fs.close();}public static FileSystem getFs() throws IOException, InterruptedException {// 1. 创建Hadoop配置对象，并设置配置信息Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://192.168.121.100:9000");conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");// 2. 获取文件系统实例，设置文件路径return FileSystem.get(URI.create(""), conf, "root");}}

执行结果如下图所示：

4. 创建文件

4.1 创建文件并写入数据

可以使用create()方法来创建文件并写入数据。

package org.example;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;import java.io.IOException;
import java.net.URI;public class Main {public static void main(String[] args) throws IOException, InterruptedException {FileSystem fs = getFs();// 创建文件/test/demo/text.txt，并向文件中写入数据FSDataOutputStream fsDataOutputStream = fs.create(new Path("/test/demo/text.txt"));String dataStr = "这是写入文件/test/demo/text.txt的示例数据\n";fsDataOutputStream.write(dataStr.getBytes());System.out.println("===============创建文件并写入数据成功===============");// 关闭文件系统fs.close();}public static FileSystem getFs() throws IOException, InterruptedException {// 1. 创建Hadoop配置对象，并设置配置信息Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://192.168.121.100:9000");conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");// 2. 获取文件系统实例，设置文件路径return FileSystem.get(URI.create(""), conf, "root");}}

4.2 创建新空白文件

可以使用createNewFile()方法来创建新空白文件。

package org.example;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;import java.io.IOException;
import java.net.URI;public class Main {public static void main(String[] args) throws IOException, InterruptedException {FileSystem fs = getFs();// 创建新空白文件/test/demo/text01.txtboolean newFile = fs.createNewFile(new Path("/test/demo/text01.txt"));if (newFile) {System.out.println("===============创建新空白文件成功===============");} else {System.out.println("===============创建新空白文件失败===============");}// 关闭文件系统fs.close();}public static FileSystem getFs() throws IOException, InterruptedException {// 1. 创建Hadoop配置对象，并设置配置信息Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://192.168.121.100:9000");conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");// 2. 获取文件系统实例，设置文件路径return FileSystem.get(URI.create(""), conf, "root");}}

5. 查看文件内容

可以使用open()方法来打开文件并读取文件内容。

package org.example;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;public class Main {public static void main(String[] args) throws IOException, InterruptedException {FileSystem fs = getFs();// 打开文件/test/demo/text.txt，并读取文件内容FSDataInputStream open = fs.open(new Path("/test/demo/text.txt"));BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(open));System.out.println("===============文件内容如下：===============");String line;while ((line = bufferedReader.readLine()) != null) {System.out.println(line);}// 关闭文件系统fs.close();}public static FileSystem getFs() throws IOException, InterruptedException {// 1. 创建Hadoop配置对象，并设置配置信息Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://192.168.121.100:9000");conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");// 2. 获取文件系统实例，设置文件路径return FileSystem.get(URI.create(""), conf, "root");}}

执行结果如下图所示：

6. 查看目录下的文件或目录信息

6.1 查看指定目录下的文件或目录信息

可以使用listStatus()方法来查看目录下的文件或目录信息。

package org.example;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;import java.io.IOException;
import java.net.URI;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;public class Main {public static void main(String[] args) throws IOException, InterruptedException {FileSystem fs = getFs();// 查看目录/test下的文件或目录信息FileStatus[] fileStatuses = fs.listStatus(new Path("/test"));for (FileStatus fileStatus : fileStatuses) {System.out.println("===============文件或目录信息如下：===============");FsPermission permission = fileStatus.getPermission();short replication = fileStatus.getReplication();String owner = fileStatus.getOwner();String group = fileStatus.getGroup();long size = fileStatus.getLen();long blockSize = fileStatus.getBlockSize();LocalDateTime localDateTime = Instant.ofEpochMilli(fileStatus.getAccessTime()).atZone(ZoneOffset.systemDefault()).toLocalDateTime();String accessTime = localDateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));Path path = fileStatus.getPath();System.out.println("权限：" + permission + "\t" + "副本数：" + replication + "\t" + "所有者：" + owner + "\t" + "群组：" + group);System.out.println("大小：" + size + "\t" + "块大小：" + blockSize + "\t" + "访问时间：" + accessTime + "\t" + "路径：" + path);}// 关闭文件系统fs.close();}public static FileSystem getFs() throws IOException, InterruptedException {// 1. 创建Hadoop配置对象，并设置配置信息Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://192.168.121.100:9000");conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");// 2. 获取文件系统实例，设置文件路径return FileSystem.get(URI.create(""), conf, "root");}}

执行结果如下图所示：

6.2 递归查看指定目录下的所有文件信息

可以使用listFiles()方法来递归查看指定目录下的所有文件信息。

package org.example;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.permission.FsPermission;import java.io.IOException;
import java.net.URI;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;public class Main {public static void main(String[] args) throws IOException, InterruptedException {FileSystem fs = getFs();// 递归查看目录/test下的所有文件信息RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/test"), true);while (listFiles.hasNext()) {FileStatus fileStatus = listFiles.next();System.out.println("===============文件或目录信息如下：===============");FsPermission permission = fileStatus.getPermission();short replication = fileStatus.getReplication();String owner = fileStatus.getOwner();String group = fileStatus.getGroup();long size = fileStatus.getLen();long blockSize = fileStatus.getBlockSize();LocalDateTime localDateTime = Instant.ofEpochMilli(fileStatus.getAccessTime()).atZone(ZoneOffset.systemDefault()).toLocalDateTime();String accessTime = localDateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));Path path = fileStatus.getPath();System.out.println("权限：" + permission + "\t" + "副本数：" + replication + "\t" + "所有者：" + owner + "\t" + "群组：" + group);System.out.println("大小：" + size + "\t" + "块大小：" + blockSize + "\t" + "访问时间：" + accessTime + "\t" + "路径：" + path);}// 关闭文件系统fs.close();}public static FileSystem getFs() throws IOException, InterruptedException {// 1. 创建Hadoop配置对象，并设置配置信息Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://192.168.121.100:9000");conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");// 2. 获取文件系统实例，设置文件路径return FileSystem.get(URI.create(""), conf, "root");}}

执行结果如下图所示：

6.3 查看指定路径信息

可以使用getFileStatus()方法来查看指定路径信息。

package org.example;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;import java.io.IOException;
import java.net.URI;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;public class Main {public static void main(String[] args) throws IOException, InterruptedException {FileSystem fs = getFs();// 查看目录/test的信息FileStatus fileStatus = fs.getFileStatus(new Path("/test"));System.out.println("===============路径信息如下：===============");FsPermission permission = fileStatus.getPermission();short replication = fileStatus.getReplication();String owner = fileStatus.getOwner();String group = fileStatus.getGroup();long size = fileStatus.getLen();long blockSize = fileStatus.getBlockSize();LocalDateTime localDateTime = Instant.ofEpochMilli(fileStatus.getAccessTime()).atZone(ZoneOffset.systemDefault()).toLocalDateTime();String accessTime = localDateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));Path path = fileStatus.getPath();System.out.println("权限：" + permission + "\t" + "副本数：" + replication + "\t" + "所有者：" + owner + "\t" + "群组：" + group);System.out.println("大小：" + size + "\t" + "块大小：" + blockSize + "\t" + "访问时间：" + accessTime + "\t" + "路径：" + path);// 关闭文件系统fs.close();}public static FileSystem getFs() throws IOException, InterruptedException {// 1. 创建Hadoop配置对象，并设置配置信息Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://192.168.121.100:9000");conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");// 2. 获取文件系统实例，设置文件路径return FileSystem.get(URI.create(""), conf, "root");}}

执行结果如下图所示：

7. 文件上传

可以使用copyFromLocalFile()方法从本地上传文件到hdfs。

package org.example;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;import java.io.IOException;
import java.net.URI;public class Main {public static void main(String[] args) throws IOException, InterruptedException {FileSystem fs = getFs();// 把本地文件D:\shiyan.csv上传到hdfs的/test/demo目录，上传后不删除本地文件D:\shiyan.csvfs.copyFromLocalFile(new Path("D:\\shiyan.csv"), new Path("/test/demo"));// 把本地文件D:\shiyan.csv上传到hdfs的/test/demo目录，上传后不删除本地文件D:\shiyan.csv，且如果hdfs上存在相同的文件则覆盖// fs.copyFromLocalFile(false, true, new Path("D:\\shiyan.csv"), new Path("/test/demo"));System.out.println("===============文件上传成功===============");// 关闭文件系统fs.close();}public static FileSystem getFs() throws IOException, InterruptedException {// 1. 创建Hadoop配置对象，并设置配置信息Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://192.168.121.100:9000");conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");// 2. 获取文件系统实例，设置文件路径return FileSystem.get(URI.create(""), conf, "root");}}

8. 向文件追加内容

可以使用append()方法追加内容到hdfs上的指定文件。

package org.example;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;import java.io.IOException;
import java.net.URI;public class Main {public static void main(String[] args) throws IOException, InterruptedException {FileSystem fs = getFs();// 向hdfs上文件/test/demo/text01.txt写入数据FSDataOutputStream fsDataOutputStream = fs.append(new Path("/test/demo/text01.txt"));// 把9,7,6,7,9写入文件/test/demo/text01.txt并换行fsDataOutputStream.write("9,7,6,7,9\n".getBytes());// 把缓冲区数据刷入文件fsDataOutputStream.flush();// 关闭资源fsDataOutputStream.close();System.out.println("===============追加数据到文件成功===============");// 关闭文件系统fs.close();}public static FileSystem getFs() throws IOException, InterruptedException {// 1. 创建Hadoop配置对象，并设置配置信息Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://192.168.121.100:9000");conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");// 2. 获取文件系统实例，设置文件路径return FileSystem.get(URI.create(""), conf, "root");}}

9. 文件下载

可以使用copyToLocalFile()方法从hdfs下载文件到本地。

package org.example;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;import java.io.IOException;
import java.net.URI;public class Main {public static void main(String[] args) throws IOException, InterruptedException {FileSystem fs = getFs();// 把hdfs上文件/test/demo/text01.txt下载到本地D:\目录fs.copyToLocalFile(new Path("/test/demo/text01.txt"), new Path("D:\\"));System.out.println("===============文件下载成功===============");// 关闭文件系统fs.close();}public static FileSystem getFs() throws IOException, InterruptedException {// 1. 创建Hadoop配置对象，并设置配置信息Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://192.168.121.100:9000");conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");// 2. 获取文件系统实例，设置文件路径return FileSystem.get(URI.create(""), conf, "root");}}

10. 移动或重命名

可以使用rename()方法来移动或重命名文件。

package org.example;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;import java.io.IOException;
import java.net.URI;public class Main {public static void main(String[] args) throws IOException, InterruptedException {FileSystem fs = getFs();// 把hdfs上文件/test/demo/text.txt重命名为/test/demo/text02.txtboolean rename = fs.rename(new Path("/test/demo/text.txt"), new Path("/test/demo/text02.txt"));if (rename) {System.out.println("===============文件重命名成功===============");} else {System.out.println("===============文件重命名失败===============");}// 把hdfs上文件/test/demo/text01.txt移动到/test/demo2目录boolean rename2 = fs.rename(new Path("/test/demo/text02.txt"), new Path("/test/demo2"));if (rename2) {System.out.println("===============文件移动成功===============");} else {System.out.println("===============文件移动失败===============");}// 关闭文件系统fs.close();}public static FileSystem getFs() throws IOException, InterruptedException {// 1. 创建Hadoop配置对象，并设置配置信息Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://192.168.121.100:9000");conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");// 2. 获取文件系统实例，设置文件路径return FileSystem.get(URI.create(""), conf, "root");}}

执行结果如下图所示：

11. 复制或重命名文件

可以使用FileUtil.copy()方法来复制或重命名文件。

package org.example;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;import java.io.IOException;
import java.net.URI;public class Main {public static void main(String[] args) throws IOException, InterruptedException {FileSystem fs = getFs();// 把hdfs上文件/test/demo/text01.txt复制到/test/demo2目录boolean copy = FileUtil.copy(fs, new Path("/test/demo/text01.txt"), fs, new Path("/test/demo2"), false, true, fs.getConf());if (copy) {System.out.println("===============复制文件成功===============");} else {System.out.println("===============复制文件失败===============");}// 把hdfs上文件/test/demo/text01.txt复制并重命名为/test/demo2/text03.txtboolean copy2 = FileUtil.copy(fs, new Path("/test/demo/text01.txt"), fs, new Path("/test/demo2/text03.txt"), false, true, fs.getConf());if (copy2) {System.out.println("===============复制文件并重命名成功===============");} else {System.out.println("===============复制文件并重命名失败===============");}// 关闭文件系统fs.close();}public static FileSystem getFs() throws IOException, InterruptedException {// 1. 创建Hadoop配置对象，并设置配置信息Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://192.168.121.100:9000");conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");// 2. 获取文件系统实例，设置文件路径return FileSystem.get(URI.create(""), conf, "root");}}

执行结果如下图所示：

12. 删除文件或目录

12.1 删除文件

可以使用delete()方法来删除文件。

package org.example;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;import java.io.IOException;
import java.net.URI;public class Main {public static void main(String[] args) throws IOException, InterruptedException {FileSystem fs = getFs();// 删除hdfs上文件/test/demo2/text03.txtboolean delete = fs.delete(new Path("/test/demo2/text03.txt"), false);if (delete) {System.out.println("===============删除文件成功===============");} else {System.out.println("===============删除文件失败===============");}// 关闭文件系统fs.close();}public static FileSystem getFs() throws IOException, InterruptedException {// 1. 创建Hadoop配置对象，并设置配置信息Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://192.168.121.100:9000");conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");// 2. 获取文件系统实例，设置文件路径return FileSystem.get(URI.create(""), conf, "root");}}

12.1 删除目录

可以使用delete()方法来删除目录。

package org.example;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;import java.io.IOException;
import java.net.URI;public class Main {public static void main(String[] args) throws IOException, InterruptedException {FileSystem fs = getFs();// 递归删除hdfs上目录/test/demo2boolean delete = fs.delete(new Path("/test/demo2"), true);if (delete) {System.out.println("===============删除目录成功===============");} else {System.out.println("===============删除目录失败===============");}// 关闭文件系统fs.close();}public static FileSystem getFs() throws IOException, InterruptedException {// 1. 创建Hadoop配置对象，并设置配置信息Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://192.168.121.100:9000");conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");// 2. 获取文件系统实例，设置文件路径return FileSystem.get(URI.create(""), conf, "root");}}

13. 检查路径是否存在

可以使用exists()方法来检查路径是否存在。

package org.example;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;import java.io.IOException;
import java.net.URI;public class Main {public static void main(String[] args) throws IOException, InterruptedException {FileSystem fs = getFs();// 检查路径/test/demo是否存在boolean exists = fs.exists(new Path("/test/demo"));if (exists) {System.out.println("===============路径存在===============");} else {System.out.println("===============路径不存在===============");}// 关闭文件系统fs.close();}public static FileSystem getFs() throws IOException, InterruptedException {// 1. 创建Hadoop配置对象，并设置配置信息Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://192.168.121.100:9000");conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");// 2. 获取文件系统实例，设置文件路径return FileSystem.get(URI.create(""), conf, "root");}}