分布式文件系统 HDFS：编程实践与 Java API 详解

这确保了程序能正确识别 HDFS 协议和配置。

3. 编写代码

新建类 MergeFile，主要逻辑包括初始化 Configuration、获取 FileSystem 实例、遍历文件列表并写入目标文件。代码如下：

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

/**
 * 过滤掉文件名满足特定条件的文件
 */
class MyPathFilter implements PathFilter {
    String reg = null;

    MyPathFilter(String reg) {
        this.reg = reg;
    }

    public boolean accept(Path path) {
        if (!(path.toString().matches(reg))) return true;
        return false;
    }
}

/**
 * 利用 FSDataOutputStream 和 FSDataInputStream 合并 HDFS 中的文件
 */
public class MergeFile {
    Path inputPath = null; // 待合并的文件所在的目录的路径
    Path outputPath = null; // 输出文件的路径

    public MergeFile(String input, String output) {
        this.inputPath = new Path(input);
        this.outputPath = new Path(output);
    }

    public void doMerge() throws IOException {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://localhost:9000");
        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        
        FileSystem fsSource = FileSystem.get(URI.create(inputPath.toString()), conf);
        FileSystem fsDst = FileSystem.get(URI.create(outputPath.toString()), conf);

        // 过滤掉输入目录中后缀为 .abc 的文件
        FileStatus[] sourceStatus = fsSource.listStatus(inputPath, new MyPathFilter(".*\\.abc"));
        FSDataOutputStream fsdos = fsDst.create(outputPath);
        PrintStream ps = new PrintStream(System.out);

        for (FileStatus sta : sourceStatus) {
            System.out.print("路径：" + sta.getPath() + " 文件大小：" + sta.getLen() + " 权限：" + sta.getPermission() + " 内容：");
            FSDataInputStream fsdis = fsSource.open(sta.getPath());
            byte[] data = new byte[1024];
            int read = -1;
            while ((read = fsdis.read(data)) > 0) {
                ps.write(data, 0, read);
                fsdos.write(data, 0, read);
            }
            fsdis.close();
        }
        ps.close();
        fsdos.close();
    }

    public static void main(String[] args) throws IOException {
        MergeFile merge = new MergeFile("hdfs://localhost:9000/user/hadoop/", "hdfs://localhost:9000/user/hadoop/merge.txt");
        merge.doMerge();
    }
}

4. 运行与验证

确保 Hadoop 集群已启动，且 HDFS 中存在测试文件（如 file1.txt, file2.txt 等）。在 Eclipse 中直接运行 main 方法。控制台会输出处理过程信息。运行结束后，可通过终端命令验证结果：

hdfs dfs -cat /user/hadoop/merge.txt

5. 部署到集群

若需在集群上独立运行，需将项目打包为 JAR 文件。在 Eclipse 中选择 Export -> Runnable JAR file，选择主类 MergeFile，勾选 "Extract required libraries into generated JAR"，导出为 HDFSExample.jar。

将 JAR 包上传至服务器后，使用 hadoop jar 命令提交任务：

hadoop jar ./myapp/HDFSExample.jar

总结

本文梳理了 HDFS 的三种主要操作方式。Shell 命令适合快速运维，Web 界面适合状态监控，而 Java API 则是构建大数据应用的基础。通过上述合并文件的实例，展示了从环境配置、依赖引入到代码编写及部署的全流程。掌握这些 API 细节，能帮助开发者更高效地处理分布式存储中的数据流转问题。