Mapreduce Combiner

Question

Ich habe einen einfachen mapreduce Code mit Mapper, Reducer und Combiner. Die Ausgabe vom Mapper wird an den Kombinierer übergeben. Aber zum Reduzierer wird die Ausgabe vom Mapper ausgegeben, anstatt vom Kombinierer ausgegeben zu werden.

Bitte helfen

Code:

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

Es scheint, dass Sie über wichtige Eigenschaft eines Kombinierers vergessen:

die Eingabearten für den Schlüssel/Wert und die Ausgabearten des Schlüssels müssen gleich sein.

Sie können keine Text/DoubleWritable aufnehmen und eine Text/Text zurückgeben. Ich schlage vor, Sie verwenden statt DoubleWritable, und tun Sie die richtige Analyse innerhalb Combiner.

Answer 2

Die Nr. 1 Regel von Combiners sind: gehen Sie nicht davon aus, dass der Combiner laufen wird. Behandeln Sie den Combiner nur als Optimierung.

Es kann nicht garantiert werden, dass der Combiner alle Ihre Daten durchläuft. In einigen Fällen, in denen die Daten nicht auf die Festplatte übertragen werden müssen, wird MapReduce den Combiner vollständig überspringen. Beachten Sie auch, dass der Combiner möglicherweise mehrfach über Teilmengen der Daten laufen kann! Es wird einmal pro Überlauf laufen.

In Ihrem Fall machen Sie diese schlechte Annahme. Sie sollten die Summe im Combiner UND im Reducer machen.

Sie sollten auch die Antwort von @ user987339 folgen. Der Eingang und Ausgang des Combiners muss identisch sein (Text, Double -> Text, Double) und muss mit dem Ausgang des Mappers und dem Eingang des Reducers übereinstimmen.

Answer 3

Combiner wird nicht immer funktionieren, wenn Sie mapreduce ausführen.

Wenn es mindestens drei Überlaufdateien gibt (die Ausgabe des Mappers wird auf die lokale Festplatte geschrieben), wird der Combiner ausgeführt, so dass die Größe der Datei reduziert werden kann, sodass sie einfach übertragen werden kann.

Die Zahl der Leckagen, für die ein Kombinierer laufen müssen, können

Answer 4

Wenn eine kombinieren Funktion verwendet wird, dann durch min.num.spills.for.combine Eigenschaft festgelegt werden, es ist die gleiche Form wie die Funktion reduzieren (und ist eine Implementierung von Reducer), außer seinen Ausgangstypen sind der Zwischenschlüssel und Werttypen (K2 und V2), so können sie die Reduce - Funktion liefern: Karte: (K1, V1) → Liste (K2, V2) kombinieren: (K2, list (V2)) → Liste (K2, V2) reduzieren: (K2, Liste (V2)) → Liste (K3, V3) Oft sind die Funktionen zum Kombinieren und Reduzieren gleich, in diesem Fall ist K3 gleich K2 und V3 ist das gleiche wie V2.