以编程方式读取 Hadoop Mapreduce 程序的输出答案

【问题标题】：Programmatically reading the output of Hadoop Mapreduce Program以编程方式读取 Hadoop Mapreduce 程序的输出
【发布时间】：2011-04-12 11:01:01
【问题描述】：

这可能是一个基本问题，但我在 Google 上找不到答案。
我有一个 map-reduce 作业，它在其输出目录中创建多个输出文件。我的 Java 应用程序在远程 hadoop 集群上执行此作业，作业完成后，它需要使用org.apache.hadoop.fs.FileSystem API 以编程方式读取输出。有可能吗？
应用程序知道输出目录，但不知道 map-reduce 作业生成的输出文件的名称。似乎没有办法以编程方式列出 hadoop 文件系统 API 中目录的内容。如何读取输出文件？
这似乎是一个司空见惯的场景，我相信它有一个解决方案。但我遗漏了一些非常明显的东西。

【问题讨论】：

标签： hadoop mapreduce hdfs

【解决方案1】：

您要查找的方法称为 listStatus(Path)。它只是将 Path 内的所有文件作为 FileStatus 数组返回。然后你可以简单地循环它们创建一个路径对象并读取它。

    FileStatus[] fss = fs.listStatus(new Path("/"));
    for (FileStatus status : fss) {
        Path path = status.getPath();
        SequenceFile.Reader reader = new SequenceFile.Reader(fs, path, conf);
        IntWritable key = new IntWritable();
        IntWritable value = new IntWritable();
        while (reader.next(key, value)) {
            System.out.println(key.get() + " | " + value.get());
        }
        reader.close();
    }

对于 Hadoop 2.x，您可以像这样设置阅读器：

 SequenceFile.Reader reader = 
           new SequenceFile.Reader(conf, SequenceFile.Reader.file(path))

【讨论】：

@Thomas, listStatus 似乎也返回其他文件，例如_成功
是的，但这不是我的问题；）你必须自己过滤
你能指导我如何过滤 _SUCCESS 文件吗？
@waqas 使用 PathFilter (hadoop.apache.org/common/docs/r0.20.2/api/org/apache/hadoop/fs/…) 和上面的 FileSystem API。
@xgMz 是的，它已被 Configuration,Reader.Option 取代。工作方式基本相同。您只需使用将路径作为参数的FileOption 为其播种。

【解决方案2】：

你有几个选择：这里有两个我有时会使用。

方法 #1： 根据您的数据大小，使用以下 HDFS 命令（找到 here，第 6 项）

hadoop fs -getmerge hdfs-output-dir local-file
// example 
hadoop fs -getmerge /user/kenny/mrjob/ /tmp/mrjob_output
// another way
hadoop fs -cat /user/kenny/mrjob/part-r-* > /tmp/mrjob_output

“这会将 HDFS 文件 hdfs-output-dir/part-* 连接到一个本地文件中。”

然后您可以只读取一个文件。（注意是本地存储，不是HDFS）

方法 #2：创建一个辅助方法：（我有一个名为 HDFS 的类，其中包含配置、文件系统实例以及其他辅助方法）

public List<Path> matchFiles(String path, final String filter) {
        List<Path> matches = new LinkedList<Path>();
        try {
            FileStatus[] statuses = fileSystem.listStatus(new Path(path), new PathFilter() {
                       public boolean accept(Path path) {
                          return path.toString().contains(filter);
                       }
                    });  
            for(FileStatus status : statuses) {
                matches.add(status.getPath());
            }
        } catch(IOException e) {
        LOGGER.error(e.getMessage(), e);
        }
        return matches;
    }

然后您可以通过如下命令调用：hdfs.matchFiles("/user/kenny/mrjob/", "part-")

【讨论】：

【解决方案3】：

            FSDataInputStream inputStream = fs.open(path);
            BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
            String record;
            while((record = reader.readLine()) != null) {
                int blankPos = record.indexOf(" ");
                System.out.println(record+"blankPos"+blankPos);
                String keyString = record.substring(0, blankPos);
                String valueString = record.substring(blankPos + 1);
                System.out.println(keyString + " | " + valueString);
            }

【讨论】：