java - procesamiento - Cambiar el nombre de los archivos de pieza en Hadoop Map Reducir

map reduce python

Question

He intentado usar la clase MultipleOutputs como en el ejemplo en la página http://hadoop.apache.org/docs/mapreduce/r0.21.0/api/index.html?org/apache/hadoop/mapreduce/lib/output/MultipleOutputs.html

Código de conductor

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

Código Reductor

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

Se encuentra que la salida del reductor cambia el nombre a text-r-00000

Pero el problema aquí es que también estoy obteniendo un archivo de parte-r-00000 vacío. ¿Es así como se espera que MultipleOutputs se comporte, o hay algún problema con mi código? Por favor aconséjame.

Otra alternativa que he probado es iterar a través de mi carpeta de salida utilizando la clase FileSystem y renombrar manualmente todos los archivos que comienzan con la parte.

¿Cuál es la mejor manera?

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

Answer 1

Esto es todo lo que necesita hacer en la clase Driver para cambiar el nombre base del archivo de salida: job.getConfiguration().set("mapreduce.output.basename", "text"); Por lo tanto, esto hará que sus archivos se llamen "text-r-00000".

Answer 2

Incluso si está utilizando MultipleOutputs , el OutputFormat predeterminado (creo que es TextOutputFormat ) todavía se está utilizando, por lo que se inicializará y creará estos archivos de part-r-xxxxx que está viendo.

El hecho de que estén vacíos se debe a que no está haciendo ningún context.write porque está utilizando MultipleOutputs . Pero eso no impide que se creen durante la inicialización.

Para deshacerse de ellos, necesita definir su OutputFormat para decir que no espera ningún resultado. Puedes hacerlo de esta manera:

job.setOutputFormat(NullOutputFormat.class);

Con ese conjunto de propiedades, esto debería garantizar que los archivos de sus partes nunca se inicien en absoluto, pero aún obtiene su resultado en las salidas MultipleOutputs .

Probablemente también podría usar LazyOutputFormat que garantizaría que los archivos de salida solo se crean cuando / si hay algunos datos, y no inicializa los archivos vacíos. Podrías hacerlo de esta manera:

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

Tenga en cuenta que está utilizando en su Reducer el prototipo MultipleOutputs.write(String namedOutput, K key, V value) , que solo utiliza una ruta de salida predeterminada que se generará en función de su namedOutput a algo como: {namedOutput}-(m|r)-{part-number} . Si desea tener más control sobre sus nombres de archivo de salida, debe usar el prototipo MultipleOutputs.write(String namedOutput, K key, V value, String baseOutputPath) que le permite obtener nombres de archivos generados en tiempo de ejecución en función de sus claves / valores.