2.3 HDFS 客户端操作与API

文档摘要

2.3 HDFS 客户端操作与API 第二章：Hadoop 分布式文件系统 HDFS (Hadoop Distributed File System) 2.3 HDFS 客户端操作与API 2.3.1 HDFS 客户端操作概述 HDFS 客户端操作是指用户或应用程序通过客户端库与 HDFS 集群进行交互，执行诸如文件和目录的创建、读取、写入、删除、重命名等操作。这些操作通常通过 HDFS 客户端 API 来实现。理解 HDFS 客户端操作是有效利用 HDFS 的基础。从用户的角度来看，HDFS 客户端操作可以大致分为以下几类：文件系统管理操作: 如创建目录、删除目录、列出目录内容、检查文件或目录是否存在、获取文件或目录的元数据等。

2.3 HDFS 客户端操作与API

第二章：Hadoop 分布式文件系统 HDFS (Hadoop Distributed File System)

2.3 HDFS 客户端操作与API

2.3.1 HDFS 客户端操作概述

HDFS 客户端操作是指用户或应用程序通过客户端库与 HDFS 集群进行交互，执行诸如文件和目录的创建、读取、写入、删除、重命名等操作。这些操作通常通过 HDFS 客户端 API 来实现。理解 HDFS 客户端操作是有效利用 HDFS 的基础。

从用户的角度来看，HDFS 客户端操作可以大致分为以下几类：

文件系统管理操作: 如创建目录、删除目录、列出目录内容、检查文件或目录是否存在、获取文件或目录的元数据等。
文件读写操作: 如创建文件、写入文件内容、读取文件内容、追加文件内容等。
文件上传下载操作: 将本地文件上传到 HDFS，或将 HDFS 文件下载到本地。
文件权限管理操作: 设置文件或目录的访问权限。

从 API 的角度来看，HDFS 客户端 API 主要围绕 org.apache.hadoop.fs 包下的类展开，其中最核心的类是 FileSystem。FileSystem 类提供了一系列方法，用于执行各种 HDFS 操作。

图 2.3.1 HDFS 客户端操作架构示意图

如上图所示，HDFS 客户端作为应用程序与 HDFS 集群之间的桥梁。应用程序通过调用 HDFS 客户端 API，例如 FileSystem 对象的方法，来发起操作请求。客户端 API 内部会与 NameNode 进行元数据交互（例如，获取文件位置信息、创建目录等），并与 DataNode 进行数据交互（例如，读取文件数据、写入文件数据）。

2.3.2 HDFS 客户端 API 详解

要使用 HDFS 客户端 API，首先需要在项目中引入 Hadoop 客户端库的依赖。如果使用 Maven，可以在 pom.xml 文件中添加以下依赖：


<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>${hadoop.version}</version> <!- -  请替换为你的 Hadoop 版本 -->
</dependency>

2.3.2.1 获取 FileSystem 对象

FileSystem 类是 HDFS 客户端 API 的核心入口点。要进行任何 HDFS 操作，首先需要获取 FileSystem 对象的实例。获取 FileSystem 对象的方式主要有两种：

1. 通过 Configuration 对象获取:

这是最常用的方式。首先创建一个 Configuration 对象，它会加载 Hadoop 的配置文件（如 core-site.xml, hdfs-site.xml 等）。然后使用 FileSystem.get(Configuration conf) 静态方法获取 FileSystem 实例。


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import java.net.URI;
public class HDFSClientExample {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        // 可以通过代码设置配置项，例如指定 NameNode 地址
        // conf.set("fs.defaultFS", "hdfs://your-namenode-host:9000");
        // 获取 FileSystem 对象 (默认会读取配置文件中的 fs.defaultFS)
        FileSystem fs = FileSystem.get(conf);
        // 或者指定 URI 获取 FileSystem 对象
        // URI uri = new URI("hdfs://your-namenode-host:9000");
        // FileSystem fs = FileSystem.get(uri, conf);
        System.out.println("FileSystem obtained: " + fs.getUri());
        fs.close(); // 关闭 FileSystem 对象，释放资源
    }
}

代码解释:

Configuration conf = new Configuration();：创建一个 Configuration 对象。Configuration 对象会加载 Hadoop 配置文件，这些文件定义了 Hadoop 集群的各种配置信息，包括 NameNode 地址、端口等。
FileSystem fs = FileSystem.get(conf);：使用 FileSystem.get(Configuration conf) 静态方法获取 FileSystem 对象。这个方法会根据 Configuration 对象中的配置信息连接到 HDFS 集群。默认情况下，它会读取配置文件中 fs.defaultFS 属性的值来确定 NameNode 的地址。
FileSystem fs = FileSystem.get(uri, conf);：另一种获取 FileSystem 对象的方式，可以显式地通过 URI 指定 NameNode 的地址。这在需要连接到不同的 HDFS 集群时非常有用。
fs.close();：使用完 FileSystem 对象后，应该调用 close() 方法关闭连接，释放资源。

2. 通过 URI 和 Configuration 对象获取:

当需要连接到特定的 HDFS 集群，或者配置文件中没有配置 fs.defaultFS 时，可以使用 FileSystem.get(URI uri, Configuration conf) 方法，通过 URI 显式指定 NameNode 的地址。


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import java.net.URI;
public class HDFSClientExample {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        URI uri = new URI("hdfs://your-namenode-host:9000"); // 替换为你的 NameNode 地址
        FileSystem fs = FileSystem.get(uri, conf);
        System.out.println("FileSystem obtained: " + fs.getUri());
        fs.close();
    }
}

代码解释:

URI uri = new URI("hdfs://your-namenode-host:9000");：创建一个 URI 对象，指定 HDFS 的协议 (hdfs) 和 NameNode 的地址 (your-namenode-host:9000)。你需要将 your-namenode-host:9000 替换为实际的 NameNode 地址和端口。
FileSystem fs = FileSystem.get(uri, conf);：使用 FileSystem.get(URI uri, Configuration conf) 方法，根据指定的 URI 和配置信息获取 FileSystem 对象。

2.3.2.2 常用 HDFS 操作 API

获取 FileSystem 对象后，就可以使用其提供的方法进行各种 HDFS 操作。以下是一些常用的 HDFS 操作 API 以及代码示例：

1. 创建目录 (mkdirs):

mkdirs(Path f) 方法用于在 HDFS 上创建目录。如果目录已存在，则不会抛出异常。如果父目录不存在，则会先创建父目录。


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
public class HDFSClientExample {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        Path dirPath = new Path("/user/hadoop/testdir"); // 要创建的目录路径
        boolean created = fs.mkdirs(dirPath);
        if (created) {
            System.out.println("Directory created successfully: " + dirPath);
        } else {
            System.out.println("Failed to create directory: " + dirPath);
        }
        fs.close();
    }
}

代码解释:

Path dirPath = new Path("/user/hadoop/testdir");：创建一个 Path 对象，表示要创建的目录路径。HDFS 的路径使用 URI 格式，例如 /user/hadoop/testdir。
boolean created = fs.mkdirs(dirPath);：调用 FileSystem 对象的 mkdirs(Path f) 方法创建目录。该方法返回一个布尔值，表示目录是否创建成功。

2. 上传文件 (copyFromLocalFile):

copyFromLocalFile(Path src, Path dst) 方法用于将本地文件上传到 HDFS。


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import java.io.File;
public class HDFSClientExample {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        Path localFilePath = new Path("localfile.txt"); // 本地文件路径 (假设当前目录下有 localfile.txt)
        Path hdfsDirPath = new Path("/user/hadoop/upload"); // HDFS 目录路径
        Path hdfsFilePath = new Path(hdfsDirPath, localFilePath.getName()); // HDFS 目标文件路径
        // 创建 HDFS 目录 (如果不存在)
        fs.mkdirs(hdfsDirPath);
        fs.copyFromLocalFile(localFilePath, hdfsFilePath);
        System.out.println("File uploaded successfully from " + localFilePath + " to " + hdfsFilePath);
        fs.close();
    }
}

代码解释:

Path localFilePath = new Path("localfile.txt");：创建 Path 对象，表示本地文件路径。
Path hdfsDirPath = new Path("/user/hadoop/upload");：创建 Path 对象，表示 HDFS 目标目录路径。
Path hdfsFilePath = new Path(hdfsDirPath, localFilePath.getName());：构建 HDFS 目标文件路径，将本地文件名作为 HDFS 目录下的文件名。
fs.mkdirs(hdfsDirPath);：确保 HDFS 目标目录存在。
fs.copyFromLocalFile(localFilePath, hdfsFilePath);：调用 FileSystem 对象的 copyFromLocalFile(Path src, Path dst) 方法上传文件。 src 参数是本地文件路径，dst 参数是 HDFS 目标文件路径。

3. 下载文件 (copyToLocalFile):

copyToLocalFile(Path src, Path dst) 方法用于将 HDFS 文件下载到本地。


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
public class HDFSClientExample {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        Path hdfsFilePath = new Path("/user/hadoop/upload/localfile.txt"); // HDFS 文件路径 (假设文件已存在)
        Path localDirPath = new Path("download"); // 本地目录路径
        Path localFilePath = new Path(localDirPath, hdfsFilePath.getName()); // 本地目标文件路径
        // 创建本地目录 (如果不存在)
        new File(localDirPath.toString()).mkdirs();
        fs.copyToLocalFile(hdfsFilePath, localFilePath);
        System.out.println("File downloaded successfully from " + hdfsFilePath + " to " + localFilePath);
        fs.close();
    }
}

代码解释:

Path hdfsFilePath = new Path("/user/hadoop/upload/localfile.txt");：创建 Path 对象，表示要下载的 HDFS 文件路径.
Path localDirPath = new Path("download");：创建 Path 对象，表示本地目标目录路径。
Path localFilePath = new Path(localDirPath, hdfsFilePath.getName());：构建本地目标文件路径，将 HDFS 文件名作为本地目录下的文件名。
new File(localDirPath.toString()).mkdirs();：确保本地目标目录存在。
fs.copyToLocalFile(hdfsFilePath, localFilePath);：调用 FileSystem 对象的 copyToLocalFile(Path src, Path dst) 方法下载文件。 src 参数是 HDFS 文件路径，dst 参数是本地目标文件路径。

4. 读取文件内容 (open, FSDataInputStream):

要读取 HDFS 文件内容，需要使用 open(Path f) 方法获取 FSDataInputStream 对象，然后像操作普通的 Java 输入流一样读取数据。


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import java.io.BufferedReader;
import java.io.InputStreamReader;
public class HDFSClientExample {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        Path hdfsFilePath = new Path("/user/hadoop/upload/localfile.txt"); // HDFS 文件路径 (假设文件已存在)
        try (FSDataInputStream inputStream = fs.open(hdfsFilePath);
             BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream))) {
            String line;
            while ((line = reader.readLine()) != null) {
                System.out.println(line);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            fs.close();
        }
    }
}

代码解释:

FSDataInputStream inputStream = fs.open(hdfsFilePath);：调用 FileSystem 对象的 open(Path f) 方法打开 HDFS 文件，返回 FSDataInputStream 对象。FSDataInputStream 是 Hadoop 提供的用于读取 HDFS 数据的输入流，它继承自 java.io.DataInputStream，并添加了一些 HDFS 特有的功能。
BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream))：为了方便按行读取文本文件，可以使用 BufferedReader 包装 FSDataInputStream。
while ((line = reader.readLine()) != null)：循环读取文件内容，每次读取一行。
try ( ... ) { ... } finally { ... }：使用 try-with-resources 语句确保 FSDataInputStream 和 FileSystem 在使用完毕后会被关闭，即使发生异常也能保证资源释放。

5. 写入文件内容 (create, FSDataOutputStream):

要向 HDFS 文件写入内容，需要使用 create(Path f) 方法获取 FSDataOutputStream 对象，然后像操作普通的 Java 输出流一样写入数据。


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.BufferedWriter;
import java.io.OutputStreamWriter;
public class HDFSClientExample {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        Path hdfsFilePath = new Path("/user/hadoop/output.txt"); // HDFS 目标文件路径
        try (FSDataOutputStream outputStream = fs.create(hdfsFilePath);
             BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(outputStream))) {
            writer.write("Hello HDFS!\n");
            writer.write("This is a test file.\n");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            fs.close();
        }
    }
}

代码解释:

FSDataOutputStream outputStream = fs.create(hdfsFilePath);：调用 FileSystem 对象的 create(Path f) 方法创建 HDFS 文件，返回 FSDataOutputStream 对象。FSDataOutputStream 是 Hadoop 提供的用于写入 HDFS 数据的输出流，它继承自 java.io.DataOutputStream，并添加了一些 HDFS 特有的功能。如果文件已存在，默认会覆盖文件内容。
BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(outputStream))：为了方便写入文本数据，可以使用 BufferedWriter 包装 FSDataOutputStream。
writer.write("Hello HDFS!\n"); 和 writer.write("This is a test file.\n");：向文件中写入数据。
try ( ... ) { ... } finally { ... }：使用 try-with-resources 语句确保 FSDataOutputStream 和 FileSystem 在使用完毕后会被关闭。

6. 列出目录内容 (listStatus):

listStatus(Path f) 方法用于列出指定 HDFS 目录下的文件和子目录的 FileStatus 对象数组。FileStatus 对象包含了文件或目录的元数据信息，如路径、大小、修改时间、权限等。


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
public class HDFSClientExample {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        Path dirPath = new Path("/user/hadoop"); // 要列出内容的目录路径
        FileStatus[] fileStatuses = fs.listStatus(dirPath);
        if (fileStatuses != null) {
            System.out.println("Contents of directory: " + dirPath);
            for (FileStatus status : fileStatuses) {
                System.out.println("- " + status.getPath() + " (type: " + (status.isDirectory() ? "directory" : "file") + ", size: " + status.getLen() + ")");
            }
        } else {
            System.out.println("Directory is empty or does not exist: " + dirPath);
        }
        fs.close();
    }
}

代码解释:

FileStatus[] fileStatuses = fs.listStatus(dirPath);：调用 FileSystem 对象的 listStatus(Path f) 方法获取目录内容，返回 FileStatus 对象数组。
for (FileStatus status : fileStatuses)：遍历 FileStatus 数组，打印每个文件或目录的路径、类型和大小等信息。
status.isDirectory()：判断 FileStatus 对象是否表示目录。
status.getLen()：获取文件大小 (对于目录，大小为 0)。
status.getPath()：获取文件或目录的 Path 对象。

7. 删除文件或目录 (delete):

delete(Path f, boolean recursive) 方法用于删除 HDFS 文件或目录。如果 recursive 参数为 true，则可以递归删除目录及其子目录和文件。


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
public class HDFSClientExample {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        Path filePathToDelete = new Path("/user/hadoop/output.txt"); // 要删除的文件路径
        Path dirPathToDelete = new Path("/user/hadoop/testdir"); // 要删除的目录路径
        boolean deletedFile = fs.delete(filePathToDelete, false); // 删除文件，非递归
        if (deletedFile) {
            System.out.println("File deleted successfully: " + filePathToDelete);
        } else {
            System.out.println("Failed to delete file: " + filePathToDelete);
        }
        boolean deletedDir = fs.delete(dirPathToDelete, true); // 删除目录，递归
        if (deletedDir) {
            System.out.println("Directory deleted successfully: " + dirPathToDelete);
        } else {
            System.out.println("Failed to delete directory: " + dirPathToDelete);
        }
        fs.close();
    }
}

代码解释:

fs.delete(filePathToDelete, false);：删除文件，recursive 参数设置为 false，表示非递归删除。对于文件删除，recursive 参数没有实际意义。
fs.delete(dirPathToDelete, true);：删除目录，recursive 参数设置为 true，表示递归删除目录及其子目录和文件。如果要删除非空目录，必须将 recursive 设置为 true。

8. 重命名文件或目录 (rename):

rename(Path src, Path dst) 方法用于重命名 HDFS 文件或目录。


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
public class HDFSClientExample {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        Path oldPath = new Path("/user/hadoop/output.txt"); // 原路径
        Path newPath = new Path("/user/hadoop/renamed_output.txt"); // 新路径
        boolean renamed = fs.rename(oldPath, newPath);
        if (renamed) {
            System.out.println("Renamed successfully from " + oldPath + " to " + newPath);
        } else {
            System.out.println("Failed to rename from " + oldPath + " to " + newPath);
        }
        fs.close();
    }
}

代码解释:

fs.rename(oldPath, newPath);：调用 FileSystem 对象的 rename(Path src, Path dst) 方法重命名文件或目录。 src 参数是原路径，dst 参数是新路径。

9. 检查文件或目录是否存在 (exists):

exists(Path f) 方法用于检查 HDFS 文件或目录是否存在。


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
public class HDFSClientExample {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        Path filePathToCheck = new Path("/user/hadoop/renamed_output.txt"); // 要检查的文件路径
        Path dirPathToCheck = new Path("/user/hadoop/upload"); // 要检查的目录路径
        boolean fileExists = fs.exists(filePathToCheck);
        if (fileExists) {
            System.out.println("File exists: " + filePathToCheck);
        } else {
            System.out.println("File does not exist: " + filePathToCheck);
        }
        boolean dirExists = fs.exists(dirPathToCheck);
        if (dirExists) {
            System.out.println("Directory exists: " + dirPathToCheck);
        } else {
            System.out.println("Directory does not exist: " + dirPathToCheck);
        }
        fs.close();
    }
}

代码解释:

fs.exists(filePathToCheck); 和 fs.exists(dirPathToCheck);：调用 FileSystem 对象的 exists(Path f) 方法检查文件或目录是否存在。该方法返回一个布尔值，表示文件或目录是否存在。

10. 获取文件元数据 (getFileStatus):

getFileStatus(Path f) 方法用于获取 HDFS 文件或目录的 FileStatus 对象，包含了文件或目录的元数据信息。


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 java.text.SimpleDateFormat;
import java.util.Date;
public class HDFSClientExample {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        Path filePath = new Path("/user/hadoop/renamed_output.txt"); // 要获取元数据的文件路径
        FileStatus fileStatus = fs.getFileStatus(filePath);
        if (fileStatus != null) {
            System.out.println("File Status for: " + filePath);
            System.out.println("- Path: " + fileStatus.getPath());
            System.out.println("- Is Directory: " + fileStatus.isDirectory());
            System.out.println("- Length: " + fileStatus.getLen());
            System.out.println("- Modification Time: " + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date(fileStatus.getModificationTime())));
            System.out.println("- Owner: " + fileStatus.getOwner());
            System.out.println("- Group: " + fileStatus.getGroup());
            System.out.println("- Permission: " + fileStatus.getPermission());
            // ... 可以获取更多元数据信息
        } else {
            System.out.println("File does not exist: " + filePath);
        }
        fs.close();
    }
}

代码解释:

FileStatus fileStatus = fs.getFileStatus(filePath);：调用 FileSystem 对象的 getFileStatus(Path f) 方法获取文件或目录的 FileStatus 对象。
fileStatus.getPath(), fileStatus.isDirectory(), fileStatus.getLen(), fileStatus.getModificationTime(), fileStatus.getOwner(), fileStatus.getGroup(), fileStatus.getPermission()：从 FileStatus 对象中获取各种元数据信息。

2.3.3 代码实践注意事项

异常处理: HDFS 客户端 API 操作可能会抛出 IOException 等异常，需要进行适当的异常处理，例如使用 try-catch 块捕获异常并进行处理。
资源释放: 使用完 FileSystem, FSDataInputStream, FSDataOutputStream 等资源后，务必调用 close() 方法关闭资源，释放连接。可以使用 try-with-resources 语句来自动管理资源释放。
配置加载: Configuration 对象会加载 Hadoop 配置文件，确保配置文件配置正确，特别是 fs.defaultFS 属性需要配置为正确的 NameNode 地址。或者在代码中显式指定 NameNode URI。
权限管理: HDFS 有权限管理机制，确保客户端用户有足够的权限执行所需的操作。
性能优化: 对于大文件读写，可以使用缓冲区 (BufferedInputStream, BufferedOutputStream) 来提高性能。可以根据实际情况调整缓冲区大小。

2.3.4 总结

本章节详细介绍了 HDFS 客户端操作与 API，包括如何获取 FileSystem 对象以及常用的 HDFS 操作 API，如创建目录、上传下载文件、读写文件内容、列出目录内容、删除文件目录、重命名、检查存在性、获取文件元数据等。通过代码示例和详细解释，读者可以掌握 HDFS 客户端 API 的基本用法，并能够在实际应用中有效地使用 HDFS 客户端 API 进行 HDFS 交互操作。掌握这些客户端操作是深入理解和应用 Hadoop 分布式文件系统 HDFS 的重要一步。