21

我尝试按照页面http://hadoop.apache.org/docs/mapreduce/r0.21.0/api/index.html?org/apache/hadoop/mapreduce/lib/output/MultipleOutputsMultipleOutputs中的示例使用该类.html

驱动程序代码

    Configuration conf = new Configuration();
    Job job = new Job(conf, "Wordcount");
    job.setJarByClass(WordCount.class);
    job.setInputFormatClass(TextInputFormat.class);
    job.setMapperClass(WordCountMapper.class);
    job.setReducerClass(WordCountReducer.class);
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(IntWritable.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
    FileInputFormat.setInputPaths(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    MultipleOutputs.addNamedOutput(job, "text", TextOutputFormat.class,
            Text.class, IntWritable.class);
    System.exit(job.waitForCompletion(true) ? 0 : 1);

减速机代码

public class WordCountReducer extends
        Reducer<Text, IntWritable, Text, IntWritable> {
    private IntWritable result = new IntWritable();
    private MultipleOutputs<Text, IntWritable> mos;
    public void setup(Context context){
        mos = new MultipleOutputs<Text, IntWritable>(context);
    }
    public void reduce(Text key, Iterable<IntWritable> values, Context context)
            throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        result.set(sum);
        //context.write(key, result);
        mos.write("text", key,result);
    }
    public void cleanup(Context context)  {
         try {
            mos.close();
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
         }
}

发现reducer的输出重命名为text-r-00000

但这里的问题是我也得到了一个空的 part-r-00000 文件。这是 MultipleOutputs 的预期行为方式，还是我的代码有问题？请指教。

我尝试过的另一种选择是使用 FileSystem 类遍历我的输出文件夹并手动重命名所有以 part 开头的文件。

什么是最好的方法？

FileSystem hdfs = FileSystem.get(configuration);
FileStatus fs[] = hdfs.listStatus(new Path(outputPath));
for (FileStatus aFile : fs) {
if (aFile.isDir()) {
hdfs.delete(aFile.getPath(), true);
// delete all directories and sub-directories (if any) in the output directory
} 
else {
if (aFile.getPath().getName().contains("_"))
hdfs.delete(aFile.getPath(), true);
// delete all log files and the _SUCCESS file in the output directory
else {
hdfs.rename(aFile.getPath(), new Path(myCustomName));
}
}
4

2 回答 2

21

即使您正在使用MultipleOutputs，默认值OutputFormat（我相信是TextOutputFormat）仍在使用中，因此它将初始化并创建part-r-xxxxx您所看到的这些文件。

它们为空的事实是因为您没有做任何事情context.write，因为您正在使用MultipleOutputs. 但这并不妨碍它们在初始化期间被创建。

要摆脱它们，您需要定义您OutputFormat说您不期望任何输出。你可以这样做：

job.setOutputFormat(NullOutputFormat.class);

使用该属性集，这应该确保您的零件文件根本不会被初始化，但您仍然可以在MultipleOutputs.

您也可以使用LazyOutputFormatwhich 来确保您的输出文件仅在/如果有一些数据时创建，而不是初始化空文件。你可以这样做：

import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat; 
LazyOutputFormat.setOutputFormatClass(job, TextOutputFormat.class);

请注意，您Reducer在原型中使用MultipleOutputs.write(String namedOutput, K key, V value)，它只使用将根据您生成的默认输出路径，namedOutput例如：{namedOutput}-(m|r)-{part-number}. 如果你想更好地控制你的输出文件名，你应该使用原型MultipleOutputs.write(String namedOutput, K key, V value, String baseOutputPath)，它可以让你在运行时根据你的键/值生成文件名。

于 2013-01-28T04:39:07.013 回答
11

这就是您在 Driver 类中更改输出文件的基本名称所需要做的所有事情： job.getConfiguration().set("mapreduce.output.basename", "text"); 因此这将导致您的文件被称为“text-r-00000”。

于 2015-02-03T12:08:55.257 回答