如何设置一个reducer to emmit< Text，IntWritable>和一个接收< Text，IntWritable>？的映射器 [英] How to set a a reducer to emmit <Text, IntWritable> and a mapper to receive <Text, IntWritable>?

问题描述

我正在使用 mapreduce 在 hadoop 上开发一些代码，使用两个映射器和两个reducer。
我一直在告诉我们使用 SequenceFileInputFormat 和 SequenceFileOutputFormat 来使第一个reducer的输出和第二个mapper的输入一起工作。
问题是我正在记录错误并在 googleing 之后我不知道为什么。

I'm developing some code on hadoop with mapreduce that uses two mappers and two reducers. I've been told to use SequenceFileInputFormat and SequenceFileOutputFormat to make the output of the first reducer and the input of the second mapper to work together. The problem is that i'm recibing an error and after googleing a lot i don't know why.

错误：

java.lang.Exception：java.io.IOException：键入地图中的键不匹配：预期 org.apache.hadoop.io。 IntWritable ，已收到 org.apache.hadoop.io。文字

java.lang.Exception: java.io.IOException: Type mismatch in key from map: expected org.apache.hadoop.io.IntWritable, received org.apache.hadoop.io.Text

键入地图中的密钥不匹配：预期
org.apache.hadoop.io。 IntWritable ，收到 org.apache.hadoop.io。文字

Type mismatch in key from map: expected org.apache.hadoop.io.IntWritable, received org.apache.hadoop.io.Text

代码：

package casoTaxis;

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.lib.input.SequenceFileInputFormat;
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.mapreduce.lib.output.SequenceFileOutputFormat;

public class Eje1{

    public static class MapperJob1 extends Mapper<Object, Text, Text, IntWritable> {
        //El metodo map recibe un conjunto clave-valor, lo procesa y lo vuelca en un contexto.adasdadada 
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            Text hackLicense; IntWritable totalAmount; //salidas
            StringTokenizer itr = new StringTokenizer(value.toString(), ",");
            itr.nextToken();
            hackLicense = new Text(itr.nextToken());
            for(int i=2; i<itr.countTokens(); i++) itr.nextToken();
            totalAmount = new IntWritable( Integer.parseInt(itr.nextToken()) );
            context.write(hackLicense, totalAmount);
        }
    }

    public static class ReducerJob1 extends Reducer<Text, IntWritable, Text, IntWritable> { //No encontre una clase InpuFormat que sea Text, IntWritable
        public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            context.write(key, new IntWritable(sum));
        }
    }

    public static class MapperJob2 extends Mapper<Text, IntWritable, Text, IntWritable> {
        //El metodo map recibe un conjunto clave-valor, lo procesa y lo vuelca en un contexto.adasdadada 
        public void map(Text key, IntWritable value, Context context) throws IOException, InterruptedException {
            context.write(key, value);
        }
    }

    public static class ReducerJob2 extends Reducer<Text, IntWritable, Text, Text> {
        public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int max = 0;
            for (IntWritable val : values) {
                int maxVal = val.get();
                if( maxVal>max ) max = maxVal;
            }
            String licencia = "Conductor con licencia = " + key;
            String recaudacion = "Recaudacion = " + max;
            context.write(new Text(licencia), new Text(recaudacion));
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf1 = new Configuration();
        Configuration conf2 = new Configuration();
        //conf2.set("mapreduce.input.keyvaluelinerecordreader.key.value.separator", " ");
        Job job1 = Job.getInstance(conf1, "Eje1-Job1");
        Job job2 = Job.getInstance(conf2, "Eje1-Job2");
        job1.setJarByClass(Eje1.class);
        job2.setJarByClass(Eje1.class);
        job1.setMapperClass(MapperJob1.class);
        job2.setMapperClass(MapperJob2.class);
        job1.setReducerClass(ReducerJob1.class);
        job2.setReducerClass(ReducerJob2.class);

        job1.setMapOutputKeyClass(Text.class);
        job1.setMapOutputValueClass(IntWritable.class);
        job1.setOutputKeyClass(Text.class);
        job1.setOutputValueClass(IntWritable.class);
        job2.setMapOutputKeyClass(Text.class);
        job2.setMapOutputKeyClass(IntWritable.class);
        job2.setOutputKeyClass(Text.class);
        job2.setOutputValueClass(Text.class);

        job1.setOutputFormatClass(SequenceFileOutputFormat.class);
        job2.setInputFormatClass(SequenceFileInputFormat.class);///asdasdads

        FileInputFormat.addInputPath(job1, new Path(args[0]));
        FileOutputFormat.setOutputPath(job1, pathIntermedio);
        FileInputFormat.addInputPath(job2, pathIntermedio);
        FileOutputFormat.setOutputPath(job2, new Path(args[1]));

        job1.waitForCompletion(true);
        System.exit(job2.waitForCompletion(true) ? 0 : 1);
    }

    private static final Path pathIntermedio = new Path("intermediate_output");

}

为什么我收到此错误？是否有更好的方法来实现这一目标？

推荐答案

错误是行

job2.setMapOutputKeyClass(Text.class);
job2.setMapOutputKeyClass(IntWritable.class);

第二个应该是：

job2.setMapOutputValueClass(IntWritable.class);

这篇关于如何设置一个reducer to emmit< Text，IntWritable>和一个接收< Text，IntWritable>？的映射器的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！