分布式文件系统 HDFS 编程实践与常用命令详解

实例：合并 HDFS 文件

假设我们需要将 HDFS 目录下符合特定规则的文件合并为一个新文件。以下是完整实现步骤。

1. 项目初始化

在 Eclipse 中创建 Java 工程，命名为 HDFSExample。确保 JDK 版本与 Hadoop 兼容（推荐 JDK 1.8+）。

2. 引入依赖

Hadoop 的 JAR 包通常位于安装目录下的 share/hadoop 子目录中。需要手动添加以下依赖到项目 Build Path：

• hadoop-common 相关 JAR 包
• hadoop-hdfs 相关 JAR 包
• 对应的 lib 子目录下的依赖

具体路径参考你的实际安装位置，例如 /usr/local/hadoop/share/hadoop/common。

3. 编写代码

创建一个名为 MergeFile 的主类。这里我们自定义了一个 MyPathFilter 来过滤掉不需要的后缀文件（如 .abc）。

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 集群已启动，且目标目录下存在测试文件。在 Eclipse 中选择 Run As -> Java Application 即可执行。

控制台会输出处理日志，若成功，可在终端验证结果：

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

5. 打包部署

生产环境中，通常需要将应用打包成 JAR 提交到集群运行。

1. 在 Eclipse 中选择 Export -> Runnable JAR file。
2. 设置主类为 MergeFile-HDFSExample。
3. 勾选 Extract required libraries into generated JAR 以包含依赖。
4. 导出为 HDFSExample.jar。

使用 hadoop jar 命令提交任务：

./bin/hadoop jar ./myapp/HDFSExample.jar

总结

HDFS 的编程接口丰富且实用。对于简单的文件管理，Shell 命令效率最高；对于需要监控的场景，Web 界面一目了然；而对于复杂的业务逻辑，Java API 提供了足够的灵活性。掌握 FileSystem 类的使用、理解流式读写机制，并结合 PathFilter 等工具，就能高效地构建基于 Hadoop 的数据处理应用。