Combiner und Reducer können anders sein?

Question

In vielen MapReduce-Programmen sehe ich einen Reduzierer, der auch als Kombinator verwendet wird. Ich weiß, dass dies an der Besonderheit dieser Programme liegt. Aber ich frage mich, ob sie anders sein können.

Answer 1

Ja, ein Combiner kann sich vom Reducer unterscheiden, obwohl Ihr Combiner immer noch die Reducer-Schnittstelle implementiert. Combiner können nur in bestimmten Fällen verwendet werden, die vom Job abhängig sind. Der Combiner funktioniert wie ein Reducer, jedoch nur auf der Teilmenge der von jedem Mapper ausgegebenen Schlüssel/Werte.

Eine Einschränkung, die Ihr Combiner im Gegensatz zu einem Reducer haben wird, ist, dass der Eingabe/Ausgabe-Schlüssel und die Werttypen den Ausgabetypen Ihres Mappers entsprechen müssen.

Answer 2

Ja, sie können sicherlich anders sein, aber ich glaube nicht, dass Sie eine andere Klasse verwenden möchten, da Sie meistens ein unerwartetes Ergebnis erhalten.

Combiner können nur für Funktionen verwendet werden, die kommutativ sind (a.b = b.a) und assoziativ {a. (B.c) = (a.b) .c}. Dies bedeutet auch, dass Combiner nur auf einer Teilmenge Ihrer Schlüssel und Werte operieren können oder gar nicht ausgeführt werden können. Trotzdem möchten Sie, dass die Ausgabe des Programms gleich bleibt.

Die Auswahl einer anderen Klasse mit anderer Logik liefert möglicherweise keine logische Ausgabe.

Answer 3

Das primäre Ziel der Kombinierer ist die Anzahl der Schlüssel-Wert-Paare zu optimieren/zu minimieren, die über das Netzwerk zwischen Mapper und Reduzierungen und damit gemischt werden , da die meisten Bandbreite wie möglich zu speichern.

Die Daumenregel Kombinierer ist es die gleichen Ein- und Ausgangsvariablentypen haben muss, ist der Grund hierfür wird Kombinierer Verwendung nicht garantiert, es kann oder nicht verwendet werden kann, in Abhängigkeit des Volumens und Anzahl der verschüttet.

Der Reduzierer kann als Kombinator verwendet werden, wenn er diese Regel erfüllt, d. H. Gleicher Ein- und Ausgang variabler Typ. Die andere wichtigste Regel für den Kombinierer ist, dass sie nur verwendet werden kann, wenn die gewünschte Funktion sowohl kommutativ als auch assoziativ ist. wie Zahlen addieren. Aber nicht im Fall wie Durchschnitt (wenn Sie gleichen Code als Reduzierer verwenden).

Nun, um Ihre Frage zu beantworten, ja natürlich können sie anders sein, und wenn Ihr Reducer verschiedene Arten von Eingabe und Ausgabevariablen hat, haben Sie keine Wahl, sondern eine andere Kopie von Ur Reducer Code zu machen und zu ändern .

Wenn Sie sich über die Logik des Reduzierers Sorgen machen, die Sie auch auf andere Weise implementieren können, sagen Sie im Falle eines Kombinierers, dass ein Sammelobjekt über einen lokalen Puffer aller Werte verfügt, die zum Kombinierer kommen Dies ist weniger riskant als die Verwendung in Reducern, da es im Fall von Reducern anfälliger ist, den Speicher zu verlieren als im Combiner. andere logische Unterschiede können sicherlich existieren und tun.

Answer 4

Hier ist die Implementierung, Sie können ohne Combiner und mit Combiner laufen, beide geben genau die gleiche Antwort. Hier hat Reducer und Combiner unterschiedliche Motive und unterschiedliche Ausführungen.

package combiner; 

import java.io.IOException; 


import org.apache.hadoop.io.LongWritable; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapreduce.Mapper; 

public class Map extends Mapper<LongWritable, Text, Text, Average> { 

Text name = new Text(); 
String[] row; 

protected void map(LongWritable offSet, Text line, Context context) throws IOException, InterruptedException { 
    row = line.toString().split(" "); 
    System.out.println("Key "+row[0]+"Value "+row[1]); 
    name.set(row[0]); 
    context.write(name, new Average(Integer.parseInt(row[1].toString()), 1)); 
}} 

reduzieren Klasse

public class Reduce extends Reducer<Text, Average, Text, LongWritable> { 
    LongWritable avg =new LongWritable(); 
    protected void reduce(Text key, Iterable<Average> val, Context context)throws IOException, InterruptedException { 
    int total=0; int count=0; long avgg=0; 

    for (Average value : val){ 
     total+=value.number*value.count; 
     count+=value.count; 
     avgg=total/count; 
     } 
    avg.set(avgg); 
    context.write(key, avg); 
} 
} 

MapObject Klasse

public class Average implements Writable { 

long number; 
int count; 

public Average() {super();} 

public Average(long number, int count) { 
    this.number = number; 
    this.count = count; 
} 

public long getNumber() {return number;} 
public void setNumber(long number) {this.number = number;} 
public int getCount() {return count;} 
public void setCount(int count) {this.count = count;} 

@Override 
public void readFields(DataInput dataInput) throws IOException { 
    number = WritableUtils.readVLong(dataInput); 
    count = WritableUtils.readVInt(dataInput);  
} 

@Override 
public void write(DataOutput dataOutput) throws IOException { 
    WritableUtils.writeVLong(dataOutput, number); 
    WritableUtils.writeVInt(dataOutput, count); 

} 
} 

Combiner Klasse

public class Combine extends Reducer<Text, Average, Text, Average>{ 

protected void reduce(Text name, Iterable<Average> val, Context context)throws IOException, InterruptedException { 
    int total=0; int count=0; long avg=0; 

    for (Average value : val){ 
     total+=value.number; 
     count+=1; 
     avg=total/count;  
     } 
    context.write(name, new Average(avg, count)); 

} 
} 

Treiberklasse

public class Driver1 { 

public static void main(String[] args) throws Exception { 

    Configuration conf = new Configuration(); 
    if (args.length != 2) { 
     System.err.println("Usage: SecondarySort <in> <out>"); 
     System.exit(2); 
    } 
    Job job = new Job(conf, "CustomCobiner"); 
    job.setJarByClass(Driver1.class); 
    job.setMapperClass(Map.class); 
    job.setCombinerClass(Combine.class); 
    job.setMapOutputKeyClass(Text.class); 
    job.setMapOutputValueClass(Average.class); 
    job.setReducerClass(Reduce.class); 
    job.setOutputKeyClass(Text.class); 
    job.setOutputValueClass(IntWritable.class);  
    FileInputFormat.addInputPath(job, new Path(args[0])); 
    FileOutputFormat.setOutputPath(job, new Path(args[1])); 
    System.exit(job.waitForCompletion(true) ? 0 : 1); 
} 
} 

Git der Code von here

ur Anregungen hinterlassen ..