Mapreduce Combiner

Question

Ho un codice mapreduce semplice con mapper, riduttore e combinatore. L'output dal mapper viene passato al combinatore. Ma al riduttore, invece dell'output dal combinatore, viene passato l'output dal mapper.

aiuto gentile

Codice:

package Combiner; 
import java.io.IOException; 
import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.io.DoubleWritable; 
import org.apache.hadoop.io.LongWritable; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapreduce.Job; 
import org.apache.hadoop.mapreduce.Mapper; 
import org.apache.hadoop.mapreduce.Reducer; 
import org.apache.hadoop.mapreduce.Mapper.Context; 
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 
import org.apache.hadoop.util.GenericOptionsParser; 

public class AverageSalary 
{ 
public static class Map extends Mapper<LongWritable, Text, Text, DoubleWritable> 
{ 
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException 
    {  
     String[] empDetails= value.toString().split(","); 
     Text unit_key = new Text(empDetails[1]);  
     DoubleWritable salary_value = new DoubleWritable(Double.parseDouble(empDetails[2])); 
     context.write(unit_key,salary_value);  

    } 
} 
public static class Combiner extends Reducer<Text,DoubleWritable, Text,Text> 
{ 
    public void reduce(final Text key, final Iterable<DoubleWritable> values, final Context context) 
    { 
     String val; 
     double sum=0; 
     int len=0; 
     while (values.iterator().hasNext()) 
     { 
      sum+=values.iterator().next().get(); 
      len++; 
     } 
     val=String.valueOf(sum)+":"+String.valueOf(len); 
     try { 
      context.write(key,new Text(val)); 
     } catch (IOException e) { 
      // TODO Auto-generated catch block 
      e.printStackTrace(); 
     } catch (InterruptedException e) { 
      // TODO Auto-generated catch block 
      e.printStackTrace(); 
     } 
    } 
} 
public static class Reduce extends Reducer<Text,Text, Text,Text> 
{ 
    public void reduce (final Text key, final Text values, final Context context) 
    { 
     //String[] sumDetails=values.toString().split(":"); 
     //double average; 
     //average=Double.parseDouble(sumDetails[0]); 
     try { 
      context.write(key,values); 
     } catch (IOException e) { 
      // TODO Auto-generated catch block 
      e.printStackTrace(); 
     } catch (InterruptedException e) { 
      // TODO Auto-generated catch block 
      e.printStackTrace(); 
     } 
    } 
} 
public static void main(String args[]) 
{ 
    Configuration conf = new Configuration(); 
    try 
    { 
    String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();  
    if (otherArgs.length != 2) {  
     System.err.println("Usage: Main <in> <out>");  
     System.exit(-1); }  
    Job job = new Job(conf, "Average salary");  
    //job.setInputFormatClass(KeyValueTextInputFormat.class);  
    FileInputFormat.addInputPath(job, new Path(otherArgs[0]));  
    FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));  
    job.setJarByClass(AverageSalary.class);  
    job.setMapperClass(Map.class);  
    job.setCombinerClass(Combiner.class); 
    job.setReducerClass(Reduce.class);  
    job.setOutputKeyClass(Text.class);  
    job.setOutputValueClass(Text.class);  

     System.exit(job.waitForCompletion(true) ? 0 : -1); 
    } catch (ClassNotFoundException e) { 
     // TODO Auto-generated catch block 
     e.printStackTrace(); 
    } catch (IOException e) { 
     // TODO Auto-generated catch block 
     e.printStackTrace(); 
    } catch (InterruptedException e) { 
     // TODO Auto-generated catch block 
     e.printStackTrace(); 
    } 
} 

}

Answer 1

Sembra che ti sei dimenticato importante proprietà di un combinatore:

i tipi di input per la chiave/valore e i tipi di uscita della chiave/valore devono essere uguali.

Non è possibile prendere in un Text/DoubleWritable e restituire un Text/Text. Ti suggerisco di utilizzare Text invece di DoubleWritable e di eseguire l'analisi corretta all'interno di Combiner.

Answer 2

La regola numero 1 di Combinatori è: non presumere che il combinatore eseguirà. Trattare il combinatore solo come ottimizzazione.

Non è garantito che il Combin esegua tutti i dati. In alcuni casi, quando i dati non devono essere versati su disco, MapReduce salterà completamente l'utilizzo del Combinatore. Nota anche che il Combiner può essere eseguito più volte su sottoinsiemi di dati! Funzionerà una volta per spandimento.

Nel tuo caso, stai facendo questa cattiva ipotesi. Dovresti fare la somma in Combiner AND the Reducer.

Inoltre, si dovrebbe seguire anche la risposta di @ user987339. L'input e l'output del combinatore devono essere identici (Testo, Doppio -> Testo, Doppio) e devono corrispondere con l'output del Mapper e l'input del Riduttore.

Answer 3

Combiner non funziona sempre quando si esegue mapreduce.

Se sono presenti almeno tre file di sversamento (output del programma di scrittura scritto su disco locale) il combinatore verrà eseguito in modo che la dimensione del file possa essere ridotta in modo che possa essere facilmente trasferita per ridurre il nodo.

Il numero di fuoriuscite per cui un combinatore necessario eseguire può essere impostato attraverso min.num.spills.for.combine proprietà

Answer 4

Se una funzione combinare viene utilizzato, allora è la stessa forma come la funzione di ridurre (ed è un'implementazione del riduttore), tranne i tipi di output sono la chiave intermedia e i tipi di valore (K2 e V2), quindi possono alimentare la funzione di riduzione: mappa: (K1, V1) → elenco (K2, V2) combinare: (K2, elenco (V2)) → lista (K2, V2) riduci: (K2, lista (V2)) → lista (K3, V3) Spesso le funzioni di combinazione e riduzione sono le stesse, nel qual caso, K3 è lo stesso di K2 e V3 è uguale a V2.