当文件格式为自定义格式时，Hadoop MultipleOutputs 不会写入多个文件答案

【问题标题】：Hadoop MultipleOutputs does not write to multiple files when file formats are custom format当文件格式为自定义格式时，Hadoop MultipleOutputs 不会写入多个文件
【发布时间】：2012-10-19 19:34:18
【问题描述】：

我正在尝试使用 MultipleOutputs api（Hadoop 版本 1.0.3）从 cassandra 读取并写入减速器输出到多个输出文件。在我的例子中，文件格式是扩展 FileOutputFormat 的自定义输出格式。我已经按照MultipleOutputs api 中所示的类似方式配置了我的工作。但是，当我运行该作业时，我只得到一个名为 part-r-0000 的输出文件，它是文本输出格式。如果job.setOutputFormatClass() 未设置，默认情况下它认为 TextOutputFormat 是格式。此外，它只允许初始化两个格式类之一。它完全忽略了我在MulitpleOutputs.addNamedOutput(job, "format1", MyCustomFileFormat1.class, Text.class, Text.class) and MulitpleOutputs.addNamedOutput(job, "format2", MyCustomFileFormat2.class, Text.class, Text.class) 中指定的输出格式。其他人是否面临类似问题或我做错了什么？

我还尝试编写一个非常简单的 MR 程序，它从文本文件中读取并以 2 种格式 TextOutputFormat 和 SequenceFileOutputFormat 写入输出，如 MultipleOutputs api 所示。但是，那里也没有运气。我只得到 1 个文本输出格式的输出文件。

有人可以帮我解决这个问题吗？

Job job = new Job(getConf(), "cfdefGen");
job.setJarByClass(CfdefGeneration.class);

//read input from cassandra column family
ConfigHelper.setInputColumnFamily(job.getConfiguration(), KEYSPACE, COLUMN_FAMILY);
job.setInputFormatClass(ColumnFamilyInputFormat.class);
job.getConfiguration().set("cassandra.consistencylevel.read", "QUORUM");

//thrift input job configurations
ConfigHelper.setInputRpcPort(job.getConfiguration(), "9160");
ConfigHelper.setInputInitialAddress(job.getConfiguration(), HOST);
ConfigHelper.setInputPartitioner(job.getConfiguration(), "RandomPartitioner");

SlicePredicate predicate = new SlicePredicate().setColumn_names(Arrays.asList(ByteBufferUtil.bytes("classification")));
//ConfigHelper.setRangeBatchSize(job.getConfiguration(), 2048);
ConfigHelper.setInputSlicePredicate(job.getConfiguration(), predicate);

//specification for mapper
job.setMapperClass(MyMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);

//specifications for reducer (writing to files)
job.setReducerClass(ReducerToFileSystem.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
//job.setOutputFormatClass(MyCdbWriter1.class);
job.setNumReduceTasks(1);

//set output path for storing output files
Path filePath = new Path(OUTPUT_DIR);
FileSystem hdfs = FileSystem.get(getConf());
if(hdfs.exists(filePath)){
    hdfs.delete(filePath, true);
}
MyCdbWriter1.setOutputPath(job, new Path(OUTPUT_DIR));

MultipleOutputs.addNamedOutput(job, "cdb1', MyCdbWriter1.class, Text.class, Text.class);
MultipleOutputs.addNamedOutput(job, "cdb2", MyCdbWriter2.class, Text.class, Text.class);

boolean success = job.waitForCompletion(true);
return success ? 0:1;

public static class ReducerToFileSystem extends Reducer<Text, Text, Text, Text>
{
    private MultipleOutputs<Text, Text> mos;

    public void setup(Context context){
        mos = new MultipleOutputs<Text, Text>(context);
    }

    //public void reduce(Text key, Text value, Context context) 
    //throws IOException, InterruptedException (This was the mistake, changed the signature and it worked fine)
    public void reduce(Text key, Iterable<Text> values, Context context)
    throws IOException, InterruptedException
    {
        //context.write(key, value);
        mos.write("cdb1", key, value, OUTPUT_DIR+"/"+"cdb1");
        mos.write("cdb2", key, value, OUTPUT_DIR+"/"+"cdb2");
        context.progress();
    }

    public void cleanup(Context context) throws IOException, InterruptedException {
        mos.close();
    }
}

public class MyCdbWriter1<K, V> extends FileOutputFormat<K, V> 
{
    @Override
    public RecordWriter<K, V> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException 
    {
    }

    public static void setOutputPath(Job job, Path outputDir) {
        job.getConfiguration().set("mapred.output.dir", outputDir.toString());
    }

    protected static class CdbDataRecord<K, V> extends RecordWriter<K, V>
    {
        @override
        write()
        close()
    }
}

【问题讨论】：

您的每种自定义输出格式需要什么配置？
我需要CDB 输出格式文件作为我的输出文件。因此，我扩展了 FileOutputFormat 并覆盖了 getRecordWriter。我通过扩展 RecordWriter 实现了我自己的写入器，它以 CDB 格式写入。我回答你的问题了吗？
那么 MyCdbWriter1 是一个带有自己的记录写入器的 OutputFormat 吗？当你运行一个只有这样一个输出的 reducer 时，你是否必须指定 outPath 以外的任何内容？
没错。当我在没有 MultipleOutputs 的情况下运行时，我指定 job.setOutputFormatClass(MyCdbWriter1.class) 效果很好。
在我回答解决方法之前，我注意到您指定的 baseFilename 不是完全限定的 HDFS（或其他）文件名。尝试在两个 addOutput 调用中使用完全限定名称，看看会发生什么。

标签： hadoop mapreduce

【解决方案1】：

我在调试后发现我的reduce方法从未被调用过。我发现我的函数定义与 API 的定义不匹配，将其从 public void reduce(Text key, Text value, Context context) 更改为 public void reduce(Text key, Iterable<Text> values, Context context)。我不知道为什么reduce方法没有@Override标签，它可以防止我的错误。

【讨论】：

【解决方案2】：

我也遇到了类似的问题 - 我的问题是我在 Map 进程中过滤了所有记录，因此没有任何内容传递给 Reduce。在 reduce 任务中使用未命名的多个输出，这仍然会产生一个 _SUCCESS 文件和一个空的 part-r-00000 文件。

【讨论】：