Multiple Output file per Hadoop Streaming con Python Mapper

Question

Sto cercando un po 'di chiarimenti sulle risposte a questa domanda qui:

Generating Separate Output files in Hadoop Streaming

mio caso d'uso è la seguente:

ho un processo mapreduce solo mappa che prende un file di input, esegue parsing e munging e quindi scrive nuovamente. Tuttavia, alcune righe potrebbero non essere in un formato errato e, in tal caso, vorrei scrivere la riga originale in un file separato.

Sembra che un modo per farlo sarebbe quello di anteporre il nome del file alla riga che sto stampando e utilizzare il parametro multipleOutputFormat. Per esempio, se inizialmente avevo:

if line_is_valid(line): 
    print name + '\t' + comments 

ho potuto invece fare:

if line_is_valid(line): 
    print valid_file_name + '\t' + name + '\t' + comments 
else: 
    print err_file_name + '\t' + line 

L'unico problema che ho con questa soluzione è che io non voglio che il nome file di apparire come la prima colonna nei file di testo. Suppongo che potrei quindi eseguire un altro lavoro per rimuovere la prima colonna di ogni file, ma sembra una cosa sciocca. Quindi:

1) È questo il modo corretto di gestire più file di output con un lavoro mapreduce python?

2) Qual è il modo migliore per sbarazzarsi di quella colonna iniziale?

Answer 1

Si può fare qualcosa come il seguente, ma comporta una piccola compilazione Java, che penso non dovrebbe essere un problema, se si desidera comunque eseguire il caso d'uso con Python- Da Python, per quanto ne so non è direttamente possibile saltare il nome file dall'output finale come richiesto dal caso d'uso in un singolo lavoro. Ma ciò che è mostrato qui sotto può renderlo possibile con facilità!

Ecco la classe Java che ha bisogno di compilato -

package com.custom; 
import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapred.lib.MultipleTextOutputFormat; 

public class CustomMultiOutputFormat extends MultipleTextOutputFormat<Text, Text> { 
     /** 
     * Use they key as part of the path for the final output file. 
     */ 
     @Override 
     protected String generateFileNameForKeyValue(Text key, Text value, String leaf) { 
      return new Path(key.toString(), leaf).toString(); 
     } 

     /** 
     * We discard the key as per your requirement 
     */ 
     @Override 
     protected Text generateActualKey(Text key, Text value) { 
      return null; 
     } 
} 

Procedura per compilare:

1. Salvare il testo in un file esattamente (senza nome) CustomMultiOutputFormat.java

2. Mentre ci si trova nella directory in cui si trova il file salvato sopra, digitare -

   $JAVA_HOME/bin/javac -cp $(hadoop classpath) -d . CustomMultiOutputFormat.java

3. Assicurarsi JAVA_HOME sia impostata/path/to/la vostra/SUNJDK prima di tentare il comando precedente.

4. Fai la tua abitudine.file jar utilizzando (tipo esattamente) -

   $JAVA_HOME/bin/jar cvf custom.jar com/custom/CustomMultiOutputFormat.class

5. Infine, eseguire il tuo lavoro come -

   hadoop jar /path/to/your/hadoop-streaming-*.jar -libjars custom.jar -outputformat com.custom.CustomMultiOutputFormat -file your_script.py -input inputpath --numReduceTasks 0 -output outputpath -mapper your_script.py

Dopo aver fatto questi si dovrebbe vedere due directory all'interno della vostra OutputPath uno con nome_file valido e altro con err_file_name. Tutti i record che hanno valid_file_name come tag andranno nella directory nome_file_valido e tutti i record che hanno nome_file_arresto andranno nella directory nome_file_errore.

Spero che tutto ciò abbia senso.