java - 如何使用 MultipleOutputs 类在 Hadoop 中输出具有特定扩展名(如 .csv)的文件

Question

我目前有一个 MapReduce 程序，它使用 MultipleOutputs 将结果输出到多个文件中。 reducer 看起来像这样:

private MultipleOutputs mo = new MultipleOutputs<NullWritable, Text>(context);
...
public void reduce(Edge keys, Iterable<NullWritable> values, Context context)
            throws IOException, InterruptedException {
        String date = records.formatDate(millis);
        out.set(keys.get(0) + "\t" + keys.get(1));
        parser.parse(key); 
        String filePath = String.format("%s/part", parser.getFileID());
        mo.write(noval, out, filePath);
    }

这与 Hadoop:权威指南 一书中的示例非常相似 - 但是，问题在于它将文件输出为纯文本。我希望将我的文件输出为 .csv 文件，但未能在书中或在线找到相关解释。

如何做到这一点？

Answer 1

您是否尝试在驱动程序中的 Job 对象完成后遍历输出文件夹以重命名文件？

只要您在 reducer 中发出(文本应该是 csv 中的行，值由分号分隔或您需要的任何内容)，您可以尝试这样的事情:

Job job = new Job(getConf());
//...
//your job setup, including the output config 
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(NullWritable.class);
//...
boolean success = job.waitForCompletion(true);
if (success){
    FileSystem hdfs = FileSystem.get(getConf());
    FileStatus fs[] = hdfs.listStatus(new Path(outputPath));
    if (fs != null){ 
        for (FileStatus aFile : fs) {
            if (!aFile.isDir()) {
                hdfs.rename(aFile.getPath(), new Path(aFile.getPath().toString()+".csv"));
            }
        }
    }
}