Java 操作 HDFS 详细教程：从环境配置到文件读写的完整实战指南（含代码示例）

1. Java 与 Hadoop 客户端环境配置

1.1 版本选择与依赖

选择合适的 Java 版本是确保 HDFS 客户端稳定运行的基础。通常建议使用 Java 8 或以上版本，因为大多数 Hadoop 发行版对 Java 的兼容性在这些版本上更成熟。在实际环境中，请先确认你的 Hadoop 集群版本对 Java 的最低和最高支持范围。

Hadoop 客户端依赖需要与目标集群版本匹配。无论是通过打包在应用中的方式，还是在服务器上直接使用客户端包，核心依赖通常包含 hadoop-common、hadoop-client、以及对应的 hdfs-client 模块。在 Maven/Gradle 中引入合适的坐标能确保 API 的可用性。

<dependencies>
  <dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>3.x.x</version>
  </dependency>
  <dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>3.x.x</version>
  </dependency>
</dependencies>

1.2 环境变量与路径配置

设置 JAVA_HOME、HADOOP_HOME 等环境变量，是 Java 调用与 Hadoop API 发现必要组件的关键步骤。确保这两个变量在应用启动脚本或服务器环境中可用。

将 Hadoop 客户端二进制添加到 PATH，可以让普通命令行工具如 hdfs dfs -ls 等命令顺利执行，便于测试与运维。

export JAVA_HOME=/usr/lib/jvm/java-8-openjdk
export HADOOP_HOME=/opt/hadoop
export PATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin

1.3 基本连接验证

在正式编写代码前，先进行简单的连接验证，以确保环境可用。通过简单的 Java 程序初始化 Configuration，再获取 FileSystem 对象即可验证连接。

注意：在分布式环境中，实际连接会受集群端口、Kerberos、代理等因素影响，测试前请确保集群允许客户端接入。

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import java.net.URI;

public class HdfsTest {
  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    // 根据集群实际地址设置默认 filesystem，如: hdfs://namenode:8020
    conf.set("fs.defaultFS", "hdfs://namenode:8020");

    try (FileSystem fs = FileSystem.get(new URI("hdfs://namenode:8020"), conf)) {
      System.out.println("Connected to HDFS: " + fs.getUri());
    }
  }
}

2. HDFS API 概览与工作原理

2.1 FileSystem 与 Path 的核心概念

HDFS 的对外访问通过 FileSystem 抽象实现，Path 用于定位文件和目录。FileSystem 提供统一的接口来创建、删除、打开、重命名文件，以及遍历目录。

Path对象本质上是对 HDFS 路径的封装，支持跨平台路径的解析与组合，便于在不同的文件系统之间迁移数据。

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.Configuration;

Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://namenode:8020");
FileSystem fs = FileSystem.get(conf);

// 示例：创建目录
Path dir = new Path("/user/hadoop/input");
fs.mkdirs(dir);

// 示例：路径拼接
Path file = new Path(dir, "data.txt");

2.2 FSDataInputStream 与 FSDataOutputStream

用于高效读写的字节流类，FSDataInputStream 和 FSDataOutputStream 支持数据的随机访问和按块提交，适合大文件、分块传输场景。

要点：在写入大文件时，使用缓冲区并在完成后正确关闭流，避免资源泄露与数据损坏。

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.InputStream;
import java.io.OutputStream;

FileSystem fs = FileSystem.get(conf);
Path src = new Path("/local/path/data.txt");
Path dst = new Path("/user/hadoop/input/data.txt");

// 写入
try (FSDataOutputStream out = fs.create(dst, true)) {
  // 写数据到 HDFS
  out.write("hello hdfs".getBytes());
}

// 读取
try (FSDataInputStream in = fs.open(dst)) {
  byte[] buffer = new byte[1024];
  int len;
  while ((len = in.read(buffer)) > 0) {
    // 处理数据
  }
}

3. 使用 Java 操作 HDFS 的基础 API

3.1 写入文件（创建并写入）

通过 FileSystem.create 或 FSDataOutputStream，可以向 HDFS 写入文件。务必设置重复写入策略与覆盖选项，以避免数据冲突。

要点：选择合适的缓冲策略，避免一次性写入过大导致内存压力。

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import java.io.OutputStream;

Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://namenode:8020");
try (FileSystem fs = FileSystem.get(conf);
     OutputStream out = fs.create(new Path("/user/hadoop/input/newfile.txt"), true)) {
  String content = "这是一个 put 到 HDFS 的示例文本。";
  out.write(content.getBytes("UTF-8"));
}

3.2 读取文件

读取 HDFS 文件通常使用 FileSystem.open 或通过 FSDataInputStream，可逐字节或按块读取。

要点：读取完成后关闭流，避免连接泄露。

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.FSDataInputStream;

Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://namenode:8020");
try (FileSystem fs = FileSystem.get(conf);
     FSDataInputStream in = fs.open(new Path("/user/hadoop/input/data.txt"))) {
  byte[] buffer = new byte[4096];
  int len;
  while ((len = in.read(buffer)) > 0) {
    // 处理读取的数据
  }
}

3.3 删除与遍历

删除文件或目录可以使用 FileSystem.delete，遍历目录可使用 RemoteIterator 或 FileStatus。

要点：对于递归删除请将 second 参数设为 true。

import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;

FileSystem fs = FileSystem.get(conf);

// 删除单个文件
fs.delete(new Path("/user/hadoop/input/old.txt"), false);

// 递归删除目录
fs.delete(new Path("/user/hadoop/input/old_dir"), true);

// 遍历目录
RemoteIterator it = fs.listStatusIterator(new Path("/user/hadoop/input"));
while (it.hasNext()) {
  FileStatus s = it.next();
  // 处理文件/目录信息
}

4. 本地到 HDFS 的文件上传实战

4.1 通过 Java 代码实现本地文件上传

将本地文件逐块读取并写入到 HDFS，适用于大文件传输与 ETL 场景。使用 FileSystem.copyFromLocalFile 可以简化操作，但对于细粒度控制，直接实现缓冲读写更灵活。

要点：在高并发场景下，考虑分片上传与并发写入策略，以及错误重试。

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import java.io.FileInputStream;
import java.io.InputStream;

Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://namenode:8020");
try (FileSystem fs = FileSystem.get(conf);
     InputStream in = new FileInputStream("/local/path/largefile.dat")) {
  Path dst = new Path("/user/hadoop/input/largefile.dat");
  try (org.apache.hadoop.fs.FSDataOutputStream out = fs.create(dst, true)) {
    byte[] buffer = new byte[8192];
    int len;
    while ((len = in.read(buffer)) > 0) {
      out.write(buffer, 0, len);
    }
  }
}

5. 从 HDFS 读取到本地与数据导出

5.1 将 HDFS 文件下载到本地

读取 HDFS 文件并保存为本地文件，常用于数据回滚、离线分析等场景。使用 FileSystem.copyToLocalFile 或直接写入本地输出流。

要点：确保本地目标路径有写权限，且目标路径不存在冲突。

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import java.io.FileOutputStream;
import java.io.OutputStream;

Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://namenode:8020");
try (FileSystem fs = FileSystem.get(conf);
     FSDataInputStream in = fs.open(new Path("/user/hadoop/input/data.txt"));
     OutputStream out = new FileOutputStream("/local/path/data.txt")) {
  byte[] buffer = new byte[4096];
  int len;
  while ((len = in.read(buffer)) > 0) {
    out.write(buffer, 0, len);
  }
}

5.2 导出到其他系统与数据管道

将 HDFS 数据输出到外部数据仓库、文件系统或消息队列，可以结合 ETL 流程与分布式处理框架实现。核心在于确保数据一致性和传输容错。

要点：对大文件使用分块传输、断点续传、以及幂等性处理。

6. 安全、权限与配置优化

6.1 认证与授权

在生产环境中，HDFS 的安全性通常通过 Kerberos、ACLs、以及基于 HDFS 的权限位来控制。客户端要遵从集群的安全策略，确保票据、密钥和凭证正确获取。

要点：避免以 root 权限或其他高权限账户直接访问生产数据，使用最小权限原则配置用户映射。

// 示例：在 Kerberos 环境下设置认证时，确保 JVM 启动时带有 krb5 配置与票据缓存
System.setProperty("java.security.auth.login.config", "/path/to/jaas.conf");

6.2 权限与性能优化

合理配置目录权限、用户组，以及 ACL，可以提升数据安全性与并发性能。对大规模读取场景，考虑设置缓存策略与客户端并发连接数。

要点：在客户端代码中尽量复用 FileSystem 实例，避免频繁创建与销毁连接。

7. 常见问题排查与性能建议

7.1 连接超时与 DNS 解析

网络不稳定或 DNS 解析慢，可能导致 Connection timed out 或 Unknown host 错误。排查思路包括：检查 namenode/port 设置、确保集群 DNS 解析正确，以及适当的连接超时参数。

要点：在 conf 中明确设置 fs.defaultFS 和 Namenode URI，避免硬编码造成移植困难。

Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://namenode:8020");
// 也可设置超时
conf.setInt("ipc.client.connection.max.idle", 10000);

7.2 版本兼容性与 API 变动

不同 Hadoop 版本之间的 API 可能存在细微差异（如方法签名、默认配置项）。在升级或跨集群迁移时，请查阅对应版本的开发文档，确保 API 调用与行为保持一致。

要点：尽量使用 abstraction 层的通用 API，减少对具体实现的耦合。

7.3 大文件与并发场景的性能优化

对于海量小文件的读写，建议合并为大文件再进行 I/O，避免 NameNode 的小文件问题。同时，使用并发写入、适当的缓冲区大小及批量提交策略，可以显著提升吞吐率。

要点：监控 JVM 堆内存、GC 次数以及 In-Flight I/O，调整 Java 与 Hadoop 客户端参数以达到稳定性与性能的平衡。