java - tutorial - El archivo de salida contiene la salida de Mapper en lugar de la salida Reducer

mapreduce mongodb

Question

Hola, estoy tratando de encontrar el promedio de algunos números usando la técnica de reducción de mapa en modo independiente. Tengo dos archivos de entrada. Contiene los valores file1: 25 25 25 25 25 y file2: 15 15 15 15 15 .

Mi programa funciona bien, pero el archivo de salida contiene la salida del asignador en lugar de la salida del reductor.

Aquí está mi código:

import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.io.Writable;
import java.io.*;
public class Average {
public static class SumCount implements Writable {
public int sum;
public int count;
@Override
public void write(DataOutput out) throws IOException {
out.writeInt(sum);
out.writeInt(count);
}
@Override
public void readFields(DataInput in) throws IOException {
sum = in.readInt();
count =in.readInt();
}
}
public static class TokenizerMapper extends Mapper<Object, Text, Text, Object>{
private final static IntWritable valueofkey = new IntWritable();
private Text word = new Text();
SumCount sc=new SumCount();
public void map(Object key, Text value, Context context
) throws IOException, InterruptedException {
StringTokenizer itr = new StringTokenizer(value.toString());
int sum=0;
int count=0;
int v;
while (itr.hasMoreTokens()) {
word.set(itr.nextToken());
v=Integer.parseInt(word.toString());
count=count+1;
sum=sum+v;
}
word.set("average");
sc.sum=sum;
sc.count=count;
context.write(word,sc);
}
}
public static class IntSumReducer extends Reducer<Text,Object,Text,IntWritable> {
private IntWritable result = new IntWritable();
public void reduce(Text key, Iterable<SumCount> values,Context context) throws IOException, InterruptedException {
int sum = 0;
int count=0;
int wholesum=0;
int wholecount=0;
for (SumCount val : values) {
wholesum=wholesum+val.sum;
wholecount=wholecount+val.count;
}
int res=wholesum/wholecount;
result.set(res);
context.write(key, result );
}
}
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "");
job.setJarByClass(Average.class);
job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);
job.setOutputKeyClass(Text.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(SumCount.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);
}
}

después de ejecutar el programa mi archivo de salida es así:

average Average$SumCount@434ba039
average Average$SumCount@434ba039

Answer 1

No puede usar su clase de Reductor IntSumReducer como combinador. Un combinador debe recibir y emitir los mismos tipos de clave / valor.

Así que eliminaría job.setCombinerClass(IntSumReducer.class); .

Recuerde que la salida de la combinación es la entrada a la reducción, por lo que escribir Text e IntWritable no va a funcionar.

Si los archivos de salida se veían como part-m-xxxxx , el problema anterior podría significar que solo se ejecutó la fase del mapa y se detuvo. Tus contadores confirmarían esto.

También tiene Reducer<Text,Object,Text,IntWritable> que debería ser Reducer<Text,SumCount,Text,IntWritable> .