Переименование файлов частей на карте Hadoop Уменьшить

Я попытался использовать класс MultipleOutputs в соответствии с примером на странице http://hadoop.apache.org/docs/mapreduce/r0.21.0/api/index.html?org/apache/hadoop/mapreduce/lib/output/MultipleOutputs.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();
        }
         }
}

Вывод редуктора будет переименован в text-r-00000

Но проблема здесь в том, что я также получаю пустой файл part-r-00000. Является ли это, как ожидается, будет работать MultipleOutputs, или есть некоторые проблемы с моим кодом? Пожалуйста, совет.

Еще одна альтернатива, которую я опробовал, - это перебрать мою выходную папку с помощью класса FileSystem и вручную переименовать все файлы, начиная с части.

Каков наилучший способ?

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));
}
}

Ответ 1

Даже если вы используете MultipleOutputs, по умолчанию используется OutputFormat (я считаю, что это TextOutputFormat), и поэтому он будет инициализировать и создать эти файлы part-r-xxxxx, которые вы видите.

Тот факт, что они пустые, состоит в том, что вы не выполняете никаких context.write, потому что используете MultipleOutputs. Но это не мешает им создаваться во время инициализации.

Чтобы избавиться от них, вам нужно определить свой OutputFormat, чтобы сказать, что вы не ожидаете выхода. Вы можете сделать это следующим образом:

job.setOutputFormat(NullOutputFormat.class);

С этим набором свойств это должно гарантировать, что ваши файлы деталей никогда не будут инициализированы вообще, но вы все равно получите свой вывод в MultipleOutputs.

Возможно, вы также можете использовать LazyOutputFormat, который гарантирует, что ваши выходные файлы создаются только в том случае, если/имеются некоторые данные, а не инициализировать пустые файлы. Вы можете сделать это следующим образом:

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), который может позволить вам получить имена файлов, сгенерированные во время выполнения, на основе ваших ключей/значений.

Ответ 2

Это все, что вам нужно сделать в классе Driver, чтобы изменить базовое имя выходного файла: job.getConfiguration().set("mapreduce.output.basename", "text"); Таким образом, это приведет к тому, что ваши файлы будут называться "text-r-00000".