使用MapReduce MultipleOutputs清空输出文件

Question

我在我的Reducer中使用MultipleOutputs，因为我想为每个键有单独的结果文件，但是，每个结果文件都是空的，尽管默认的结果文件部分-r-xxxx被创建并包含正确的值。

这是我的代码为JobDriver和Reducer

主类

public static void main(String[] args) throws Exception {
    int currentIteration = 0;
    int reducerCount, roundCount;
    Configuration conf = createConfiguration(currentIteration);
    cleanEnvironment(conf);
    Job job = new Job(conf, "cfim");
    //Input and output format configuration
    job.setMapperClass(TransactionsMapper.class);
    job.setReducerClass(PatriciaReducer.class);
    job.setInputFormatClass(TransactionInputFormat.class);
    job.setMapOutputKeyClass(LongWritable.class);
    job.setMapOutputValueClass(Text.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);
    reducerCount = roundCount = Math.floorDiv(getRoundCount(conf), Integer.parseInt(conf.get(MRConstants.mergeFactorSpecifier)));
    FileInputFormat.addInputPath(job, new Path("/home/cloudera/datasets/input"));
    Path outputPath = new Path(String.format(MRConstants.outputPathFormat, outputDir, currentIteration));
    FileOutputFormat.setOutputPath(job, outputPath);
    MultipleOutputs.addNamedOutput(job, "key", TextOutputFormat.class, LongWritable.class, Text.class);
    job.waitForCompletion(true);

减速机类

public class PatriciaReducer extends Reducer<LongWritable, Text, LongWritable, Text> {
private ITreeManager treeManager;
private SerializationManager serializationManager;
private MultipleOutputs<LongWritable, Text> mos;
@Override 
protected void setup(Context context) throws IOException ,InterruptedException {
    treeManager = new PatriciaTreeManager();
    serializationManager = new SerializationManager();
    mos = new MultipleOutputs<LongWritable, Text>(context);
}
@Override
protected void reduce(LongWritable key, Iterable<Text> items, Context context)
        throws IOException, InterruptedException {
    Iterator<Text> patriciaIterator = items.iterator();
    PatriciaTree tree = new PatriciaTree();
    if (patriciaIterator.hasNext()){
        Text input = patriciaIterator.next();
        tree = serializationManager.deserializePatriciaTree(input.toString());
    }
    while(patriciaIterator.hasNext()){
        Text input = patriciaIterator.next();
        PatriciaTree mergeableTree = serializationManager.deserializePatriciaTree(input.toString());
        tree = treeManager.mergeTree(tree, mergeableTree, false);
    }
    Text outputValue = new Text(serializationManager.serializeAsJson(tree));
    mos.write("key", key, outputValue, generateOutputPath(key));
    context.write(key, outputValue);
}
@Override
protected void finalize() throws Throwable {
    // TODO Auto-generated method stub
    super.finalize();
    mos.close();
}
private String generateOutputPath(LongWritable key) throws IOException {
    String outputPath = String.format("%s-%s", MRConstants.reduceResultValue, key.toString());
    return outputPath;
}   

}

我做错了什么吗?

Answer 1

我发现我使用了错误的方法来关闭多个输出对象。在清理方法而不是finalize方法中关闭MultipleOutputs后，一切都工作完美