HDFS - Java API 实现 HDFS 文件上传与下载的完整案例

👋 大家好，欢迎来到我的技术博客！
📚 在这里，我会分享学习笔记、实战经验与技术思考，力求用简单的方式讲清楚复杂的问题。
🎯 本文将围绕HDFS这个话题展开，希望能为你带来一些启发或实用的参考。
🌱 无论你是刚入门的新手，还是正在进阶的开发者，希望你都能有所收获！

HDFS - Java API 实现 HDFS 文件上传与下载的完整案例 🌐

在大数据生态系统中，Hadoop 分布式文件系统（Hadoop Distributed File System，简称 HDFS）是核心组件之一。它为海量数据提供了高吞吐量、高容错性的存储解决方案，广泛应用于日志分析、数据仓库、机器学习等领域。随着企业对数据处理能力要求的不断提升，掌握如何通过编程方式操作 HDFS 成为了开发者的必备技能。

本文将详细介绍如何使用 Java API 来实现 HDFS 的文件上传与下载功能，并提供完整的代码示例、异常处理机制、性能优化建议以及实际应用场景解析。我们还将借助 Mermaid 图表来展示 HDFS 架构和数据流过程，帮助读者更直观地理解其工作原理。

📚 什么是 HDFS？

HDFS 是 Apache Hadoop 项目中的分布式文件系统模块，设计用于运行在廉价硬件上，能够存储超大规模的数据集（TB 到 PB 级别）。它采用主从架构（Master-Slave），主要由以下几个组件构成：

• NameNode：管理文件系统的命名空间（Namespace），维护文件系统树及文件到数据块的映射关系。
• DataNode：负责存储实际的数据块，定期向 NameNode 汇报状态。
• Secondary NameNode：辅助 NameNode 进行检查点操作，防止元数据丢失（注意：不是热备节点）。

HDFS 具有以下特点：

• 高容错性：自动复制数据块以应对节点故障
• 流式数据访问：一次写入，多次读取
• 大文件支持：适合存储大尺寸文件
• 可扩展性强：可轻松扩展至数千个节点

更多关于 HDFS 的基础知识可以参考 Apache Hadoop 官方文档

🔧 开发环境准备

在开始编码之前，我们需要搭建一个可用的开发环境。虽然真实的 HDFS 集群通常部署在 Linux 服务器上，但我们可以通过本地伪分布式模式进行测试。

1. 安装 Hadoop

你可以从 Hadoop 官网 下载最新稳定版本的 Hadoop 发行包。解压后配置 hadoop-env.sh 和 core-site.xml、hdfs-site.xml 等配置文件。

如果你使用的是 Windows 系统进行开发，请确保安装了 WinUtils 工具并设置好 HADOOP_HOME 环境变量，否则可能会遇到权限或路径问题。

2. 启动 HDFS 服务

# 格式化 NameNode（首次启动时执行）
bin/hdfs namenode -format

# 启动 NameNode 和 DataNode
sbin/start-dfs.sh

启动成功后，可以通过浏览器访问 http://localhost:9870 查看 HDFS Web UI 界面（旧版本为 50070 端口）。

💻 Maven 项目结构与依赖配置

我们将使用 Maven 构建 Java 项目。创建一个新的 Maven 工程，并在 pom.xml 中添加必要的依赖项。

<dependencies>
    <!-- Hadoop Client 核心依赖 -->
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>3.3.6</version>
    </dependency>

    <!-- 日志框架 -->
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-simple</artifactId>
        <version>2.0.7</version>
    </dependency>

    <!-- JUnit 测试 -->
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.13.2</version>
        <scope>test</scope>
    </dependency>
</dependencies>

📌 注意：请根据你的 Hadoop 集群版本选择合适的 hadoop-client 版本，避免因版本不兼容导致序列化错误或连接失败。

🧩 HDFS Java API 基础概念

Hadoop 提供了一套丰富的 Java API 来操作 HDFS，核心类位于 org.apache.hadoop.fs 包下。

这些类构成了我们后续所有操作的基础。

📤 实现文件上传功能

文件上传是指将本地磁盘上的文件写入 HDFS 的指定路径。这是最常见的 HDFS 操作之一，常用于日志收集、ETL 数据导入等场景。

✅ 步骤说明

1. 创建 Configuration 实例并设置 HDFS 地址
2. 获取 FileSystem 实例
3. 打开本地文件输入流
4. 在 HDFS 上创建输出流
5. 使用工具类完成数据拷贝
6. 关闭资源

🧪 示例代码：上传单个文件

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import java.io.IOException;
import java.net.URI;

public class HDFSUploader {

    private static final String HDFS_URI = "hdfs://localhost:9000"; // 根据实际情况修改

    public static void uploadFile(String localPath, String hdfsPath) throws IOException {
        Configuration conf = new Configuration();
        // 设置默认文件系统
        conf.set("fs.defaultFS", HDFS_URI);

        FileSystem fs = null;
        FileInputStream fis = null;
        FSDataOutputStream out = null;

        try {
            fs = FileSystem.get(URI.create(HDFS_URI), conf);
            Path dstPath = new Path(hdfsPath);

            // 检查目标路径是否已存在
            if (fs.exists(dstPath)) {
                System.out.println("⚠️ 目标文件已存在，即将覆盖: " + hdfsPath);
            }

            fis = new FileInputStream(localPath);
            out = fs.create(dstPath);

            // 使用 IOUtils 拷贝数据
            IOUtils.copyBytes(fis, out, 4096, true);

            System.out.println("✅ 文件上传成功！本地路径：" + localPath + " → HDFS 路径：" + hdfsPath);

        } finally {
            IOUtils.closeStream(fis);
            IOUtils.closeStream(out);
            if (fs != null) {
                fs.close();
            }
        }
    }

    public static void main(String[] args) {
        try {
            uploadFile("/Users/demo/data/input.txt", "/user/demo/upload/input.txt");
        } catch (IOException e) {
            System.err.println("❌ 文件上传失败：" + e.getMessage());
            e.printStackTrace();
        }
    }
}

🔍 代码解析

• IOUtils.copyBytes() 是 Hadoop 提供的高效工具方法，支持指定缓冲区大小并自动关闭流（当最后一个参数为 true 时）。
• fs.exists() 用于判断 HDFS 路径是否存在，避免意外覆盖重要文件。
• 使用 try-finally 确保资源被正确释放，防止内存泄漏。

📥 实现文件下载功能

与上传相反，文件下载是将 HDFS 中的文件拉取到本地磁盘的过程，适用于数据分析结果导出、模型文件获取等场景。

✅ 步骤说明

1. 获取 FileSystem 实例
2. 检查源文件是否存在
3. 打开 HDFS 输入流
4. 创建本地输出流
5. 拷贝数据
6. 关闭资源

🧪 示例代码：下载单个文件

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;

public class HDFSDownloader {

    private static final String HDFS_URI = "hdfs://localhost:9000";

    public static void downloadFile(String hdfsPath, String localPath) throws IOException {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", HDFS_URI);

        FileSystem fs = null;
        FSDataInputStream in = null;
        FileOutputStream fos = null;

        try {
            fs = FileSystem.get(URI.create(HDFS_URI), conf);
            Path srcPath = new Path(hdfsPath);

            if (!fs.exists(srcPath)) {
                throw new IOException("⛔ HDFS 文件不存在：" + hdfsPath);
            }

            FileStatus fileStatus = fs.getFileStatus(srcPath);
            System.out.println("📄 正在下载文件，大小：" + fileStatus.getLen() + " 字节");

            in = fs.open(srcPath);
            fos = new FileOutputStream(localPath);

            IOUtils.copyBytes(in, fos, 4096, true);

            System.out.println("✅ 文件下载成功！HDFS 路径：" + hdfsPath + " → 本地路径：" + localPath);

        } finally {
            IOUtils.closeStream(in);
            IOUtils.closeStream(fos);
            if (fs != null) {
                fs.close();
            }
        }
    }

    public static void main(String[] args) {
        try {
            downloadFile("/user/demo/upload/input.txt", "/Users/demo/data/downloaded_input.txt");
        } catch (IOException e) {
            System.err.println("❌ 文件下载失败：" + e.getMessage());
            e.printStackTrace();
        }
    }
}

⚠️ 异常处理增强

我们可以进一步封装异常类型，例如区分“文件不存在”、“权限不足”、“网络中断”等情况：

if (e instanceof FileNotFoundException) {
    System.err.println("🔍 文件未找到，请检查路径是否正确");
} else if (e instanceof AccessControlException) {
    System.err.println("🔐 权限拒绝，请确认用户具有读取权限");
} else {
    System.err.println("💔 未知错误：" + e.getMessage());
}

🔄 批量上传与下载（目录级操作）

在真实业务中，往往需要处理整个目录而非单个文件。下面我们实现递归上传和下载功能。

📂 递归上传目录

public static void uploadDirectory(String localDir, String hdfsDir) throws IOException {
    Configuration conf = new Configuration();
    conf.set("fs.defaultFS", HDFS_URI);
    
    FileSystem fs = FileSystem.get(URI.create(HDFS_URI), conf);
    File dir = new File(localDir);

    if (!dir.isDirectory()) {
        throw new IllegalArgumentException("提供的本地路径不是一个目录");
    }

    Path dstPath = new Path(hdfsDir);
    if (!fs.exists(dstPath)) {
        fs.mkdirs(dstPath);
    }

    for (File file : dir.listFiles()) {
        if (file.isFile()) {
            Path dstFile = new Path(hdfsDir, file.getName());
            try (FileInputStream fis = new FileInputStream(file);
                 FSDataOutputStream out = fs.create(dstFile)) {
                IOUtils.copyBytes(fis, out, 4096, false);
                System.out.println("📤 已上传：" + file.getName());
            }
        }
    }

    fs.close();
    System.out.println("🎉 目录上传完成！");
}

📂 递归下载目录

public static void downloadDirectory(String hdfsDir, String localDir) throws IOException {
    Configuration conf = new Configuration();
    conf.set("fs.defaultFS", HDFS_URI);

    FileSystem fs = FileSystem.get(URI.create(HDFS_URI), conf);
    Path srcPath = new Path(hdfsDir);

    if (!fs.exists(srcPath) || !fs.getFileStatus(srcPath).isDirectory()) {
        throw new IOException("HDFS 路径不存在或不是目录：" + hdfsDir);
    }

    File localDirectory = new File(localDir);
    if (!localDirectory.exists()) {
        localDirectory.mkdirs();
    }

    RemoteIterator<LocatedFileStatus> files = fs.listFiles(srcPath, false);
    while (files.hasNext()) {
        LocatedFileStatus file = files.next();
        Path filePath = file.getPath();
        String fileName = file.getPath().getName();
        File localFile = new File(localDirectory, fileName);

        try (FSDataInputStream in = fs.open(filePath);
             FileOutputStream fos = new FileOutputStream(localFile)) {
            IOUtils.copyBytes(in, fos, 4096, false);
            System.out.println("📥 已下载：" + fileName);
        }
    }

    fs.close();
    System.out.println("🎉 目录下载完成！");
}

📊 HDFS 数据流图解（Mermaid）

为了更好地理解文件上传过程中的数据流向，我们使用 Mermaid 绘制流程图：

该图展示了 HDFS 写入文件时典型的三副本复制流程。客户端不会直接将数据发送给所有 DataNode，而是构建一个 pipeline，依次传递数据包并等待确认，从而保证一致性和效率。

🔐 认证与安全机制

在生产环境中，HDFS 通常启用了 Kerberos 认证或其他安全策略。此时直接连接会失败。

启用 Kerberos 支持

// 加载 Kerberos 配置
System.setProperty("java.security.krb5.conf", "/etc/krb5.conf");
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://cluster-nn:8020");
conf.set("hadoop.security.authentication", "kerberos");

UserGroupInformation.setConfiguration(conf);
UserGroupInformation.loginUserFromKeytab("dev-user@EXAMPLE.COM", "/home/dev/dev.keytab");

FileSystem fs = FileSystem.get(conf); // 自动携带认证信息

📌 更多关于 Hadoop 安全的内容可参考 Cloudera 安全指南

🧹 文件系统常用操作汇总

除了上传下载，我们还经常需要进行如下操作：

列出目录内容

public static void listFiles(String hdfsPath) throws IOException {
    Configuration conf = new Configuration();
    conf.set("fs.defaultFS", HDFS_URI);

    FileSystem fs = FileSystem.get(URI.create(HDFS_URI), conf);
    Path path = new Path(hdfsPath);

    if (fs.exists(path) && fs.isDirectory(path)) {
        RemoteIterator<LocatedFileStatus> iterator = fs.listFiles(path, false);
        while (iterator.hasNext()) {
            LocatedFileStatus status = iterator.next();
            System.out.printf("📁 %-30s %10d bytes %s\n",
                    status.getPath().getName(),
                    status.getLen(),
                    new Date(status.getModificationTime()));
        }
    } else {
        System.out.println("❌ 路径不存在或不是目录");
    }

    fs.close();
}

删除文件/目录

public static boolean deleteFile(String hdfsPath, boolean recursive) throws IOException {
    Configuration conf = new Configuration();
    conf.set("fs.defaultFS", HDFS_URI);

    FileSystem fs = FileSystem.get(URI.create(HDFS_URI), conf);
    Path path = new Path(hdfsPath);

    boolean result = fs.delete(path, recursive);
    if (result) {
        System.out.println("🗑️ 成功删除：" + hdfsPath);
    } else {
        System.out.println("⚠️ 删除失败或文件不存在：" + hdfsPath);
    }

    fs.close();
    return result;
}

创建目录

public static void mkdir(String hdfsPath) throws IOException {
    Configuration conf = new Configuration();
    conf.set("fs.defaultFS", HDFS_URI);

    FileSystem fs = FileSystem.get(URI.create(HDFS_URI), conf);
    Path path = new Path(hdfsPath);

    if (fs.mkdirs(path)) {
        System.out.println("✅ 目录创建成功：" + hdfsPath);
    } else {
        System.out.println("❌ 目录创建失败（可能已存在）：" + hdfsPath);
    }

    fs.close();
}

🧪 单元测试实践

为了确保代码稳定性，建议编写 JUnit 测试用例。

import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import static org.junit.Assert.*;

public class HDFSTest {

    private Configuration conf;
    private FileSystem fs;

    @Before
    public void setUp() throws Exception {
        conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://localhost:9000");
        fs = FileSystem.get(conf);
    }

    @After
    public void tearDown() throws Exception {
        if (fs != null) {
            fs.close();
        }
    }

    @Test
    public void testUploadAndDownload() throws Exception {
        String localSrc = "/tmp/test_upload.txt";
        String hdfsDst = "/tmp/test_roundtrip.txt";
        String localBack = "/tmp/test_download.txt";

        // 准备测试文件
        java.nio.file.Files.write(java.nio.file.Paths.get(localSrc), "Hello HDFS!".getBytes());

        // 上传
        HDFSUploader.uploadFile(localSrc, hdfsDst);
        assertTrue(fs.exists(new Path(hdfsDst)));

        // 下载
        HDFSDownloader.downloadFile(hdfsDst, localBack);
        assertTrue(new java.io.File(localBack).exists());

        // 验证内容
        String content = new String(java.nio.file.Files.readAllBytes(java.nio.file.Paths.get(localBack)));
        assertEquals("Hello HDFS!", content);

        // 清理
        fs.delete(new Path(hdfsDst), false);
        new java.io.File(localSrc).delete();
        new java.io.File(localBack).delete();
    }
}

🚀 性能优化建议

在处理大文件或高频操作时，应注意以下几点以提升性能：

1. 调整缓冲区大小

默认缓冲区为 4KB，对于大文件可增大至 64KB 或更高：

conf.setInt("io.file.buffer.size", 65536); // 64KB

2. 启用短路本地读取（Short-Circuit Local Reads）

当客户端与 DataNode 在同一节点时，可绕过 TCP 直接读取本地文件：

<!-- core-site.xml -->
<property>
    <name>dfs.client.read.shortcircuit</name>
    <value>true</value>
</property>

3. 使用缓存重复连接

频繁创建 FileSystem 实例代价较高，应尽量复用：

// 正确做法：静态共享实例
private static final FileSystem FS = getFileSystemInstance();

private static FileSystem getFileSystemInstance() throws IOException {
    Configuration conf = new Configuration();
    conf.set("fs.defaultFS", HDFS_URI);
    return FileSystem.get(conf);
}

4. 并行处理多个文件

利用线程池加速批量传输：

ExecutorService executor = Executors.newFixedThreadPool(4);
List<File> files = Arrays.asList(...);

for (File f : files) {
    executor.submit(() -> {
        try {
            uploadFile(f.getAbsolutePath(), "/batch/" + f.getName());
        } catch (IOException e) {
            e.printStackTrace();
        }
    });
}

executor.shutdown();
executor.awaitTermination(1, TimeUnit.HOURS);

🌐 实际应用场景举例

场景一：日志聚合系统

每天凌晨从各台应用服务器拉取日志文件上传至 HDFS，供 Spark/Flink 进行离线分析。

// 定时任务调度
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
scheduler.scheduleAtFixedRate(LogUploader::uploadDailyLogs, 0, 1, TimeUnit.DAYS);

场景二：机器学习模型发布

训练好的模型文件（如 .pkl, .pb）上传至 HDFS，供在线服务节点统一拉取更新。

ModelDeployer.uploadModel("model_v2.pkl", "/models/current/");

场景三：数据湖接入层

构建 ETL 管道，定时将 MySQL 导出的 CSV 文件加载进 HDFS，作为数仓原始层（ODS）。

MySQLExporter.exportToCSV("SELECT * FROM orders WHERE date = CURDATE()", "/data/staging/orders.csv");
HDFSUploader.uploadFile("/data/staging/orders.csv", "/ods/orders/");

🧭 故障排查常见问题

排查技巧：始终先查看 $HADOOP_HOME/logs/ 下的日志文件，尤其是 namenode.log 和 datanode.log。

📈 HDFS 架构可视化（Mermaid）

下面是一个简化的 HDFS 架构图，展示客户端、NameNode、DataNode 之间的交互关系：

此图表明：NameNode 不存储实际数据，只管理元数据；而数据块分布在多个 DataNode 上，并保持多副本冗余。

🛡️ 最佳实践总结

1. 始终关闭资源：使用 try-with-resources 或 finally 块确保 FileSystem 被关闭
2. 合理设置副本数：通过 dfs.replication 控制可靠性与成本平衡
3. 避免小文件过多：每个文件占用 NameNode 内存，建议合并或使用 HAR 归档
4. 监控磁盘使用率：DataNode 磁盘满会导致写入失败
5. 定期格式化编辑日志：利用 Secondary NameNode 或启用 HA 避免元数据膨胀

🌱 学习资源推荐

• Hadoop 官方文档 —— 最权威的技术手册
• HDFS Architecture Guide —— 深入了解内部机制
• Cloudera Developer Training —— 免费自学课程
• Apache Hadoop Wiki —— 社区知识库

🎯 结语

通过本文的学习，你应该已经掌握了如何使用 Java API 实现 HDFS 的文件上传与下载功能，并了解了背后的原理、最佳实践和常见问题解决方案。无论是构建数据管道、实现自动化运维脚本，还是开发大数据平台前端服务，这些技能都将为你打下坚实基础。

记住：理论结合实践才是掌握技术的关键。不妨现在就动手写一段代码，把你的第一个文件传到 HDFS 上吧！🚀

“The best way to learn is by doing.” —— Unknown

Happy Coding! 💻✨

🙌 感谢你读到这里！
🔍 技术之路没有捷径，但每一次阅读、思考和实践，都在悄悄拉近你与目标的距离。
💡 如果本文对你有帮助，不妨 👍 点赞、📌 收藏、📤 分享 给更多需要的朋友！
💬 欢迎在评论区留下你的想法、疑问或建议，我会一一回复，我们一起交流、共同成长 🌿
🔔 关注我，不错过下一篇干货！我们下期再见！✨

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/qq_41187124/article/details/157684570