java - Hadoop 2( yarn )。获取 java.io.IOException : wrong key class. 异常

Question

我正在尝试运行 hadoop 2 MapReduce output_format_class 为SequenceFileOutputFormat 的过程和 input_format_class是 SequenceFileInputFormat .

我选择了Mapper以 BytesWritable 的形式发出键和值.对于Reducer它发出 key 为IntWritable和值为BytesWritable .

每次我收到以下错误:

Error: java.io.IOException: wrong key class: org.apache.hadoop.io.BytesWritable is not class org.apache.hadoop.io.IntWritable
        at org.apache.hadoop.io.SequenceFile$Writer.append(SequenceFile.java:1306)
        at org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat$1.write(SequenceFileOutputFormat.java:83)
        at org.apache.hadoop.mapred.ReduceTask$NewTrackingRecordWriter.write(ReduceTask.java:558)
        at org.apache.hadoop.mapreduce.task.TaskInputOutputContextImpl.write(TaskInputOutputContextImpl.java:89)
        at org.apache.hadoop.mapreduce.lib.reduce.WrappedReducer$Context.write(WrappedReducer.java:105)
        at org.apache.hadoop.mapreduce.Reducer.reduce(Reducer.java:150)
        at org.apache.hadoop.mapreduce.Reducer.run(Reducer.java:171)
        at org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:627)
        at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:389)
        at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:163)
        at java.security.AccessController.doPrivileged(Native Method)
        at javax.security.auth.Subject.doAs(Subject.java:415)
        at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1628)
        at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:158)

我发现当我定义 OutputFormat不像 SequenceFileOutputFormat问题已解决，但我需要它作为 SequenceFileOutputFormat .

这是主要的:

    Configuration conf = new Configuration(true);
    conf.set("refpath", "/out/Sample1/Local/EU/CloudBurst/BinaryFiles/ref.br");
    conf.set("qrypath",   "/out/Sample1/Local/EU/CloudBurst/BinaryFiles/qry.br");
    conf.set("MIN_READ_LEN",      Integer.toString(MIN_READ_LEN));
    conf.set("MAX_READ_LEN",      Integer.toString(MAX_READ_LEN));
    conf.set("K",                 Integer.toString(K));
    conf.set("SEED_LEN",          Integer.toString(SEED_LEN));
    conf.set("FLANK_LEN",         Integer.toString(FLANK_LEN));
    conf.set("ALLOW_DIFFERENCES", Integer.toString(ALLOW_DIFFERENCES));
    conf.set("BLOCK_SIZE",        Integer.toString(BLOCK_SIZE));
    conf.set("REDUNDANCY",        Integer.toString(REDUNDANCY));
    conf.set("FILTER_ALIGNMENTS", (FILTER_ALIGNMENTS ? "1" : "0"));

    Job job = new Job(conf,"CloudBurst");
    job.setNumReduceTasks(NUM_REDUCE_TASKS); // MV2

    //conf.setNumMapTasks(NUM_MAP_TASKS); TODO find solution for mv2

    FileInputFormat.addInputPath(job, new Path("/out/Sample1/Local/EU/CloudBurst/BinaryFiles/ref.br"));//TODO change it fit to the params
    FileInputFormat.addInputPath(job, new Path("/out/Sample1/Local/EU/CloudBurst/BinaryFiles/qry.br"));//TODO change it fit to the params

    job.setJarByClass(MerReduce.class);//mv2

    job.setInputFormatClass(SequenceFileInputFormat.class);
    job.setOutputFormatClass(SequenceFileOutputFormat.class);

    // The order of seeds is not important, but make sure the reference seeds are seen before the qry seeds
    job.setPartitionerClass(MerReduce.PartitionMers.class); // mv2
    job.setGroupingComparatorClass(MerReduce.GroupMersWC.class); //mv2 TODO

    job.setMapperClass(MerReduce.MapClass.class);
    job.setReducerClass(MerReduce.ReduceClass.class);

    job.setMapOutputKeyClass(BytesWritable.class);//mv2
    job.setMapOutputValueClass(BytesWritable.class);//mv2
    job.setOutputKeyClass(IntWritable.class);
    job.setOutputValueClass(BytesWritable.class);

    Path oPath = new Path("/out/Sample1/Local/EU/Vectors");//TODO change it fit to the params
    //conf.setOutputPath(oPath);
    FileOutputFormat.setOutputPath(job, oPath);
    System.err.println("  Removing old results");
    FileSystem.get(conf).delete(oPath);

    int code = job.waitForCompletion(true) ? 0 : 1;

    System.err.println("Finished");
}

映射器类标题:

public static class MapClass extends Mapper<IntWritable, BytesWritable, BytesWritable, BytesWritable>
public void map(IntWritable id, BytesWritable rawRecord,Context context) throws IOException, InterruptedException 

reducer 类标题:

public static class ReduceClass extends Reducer (BytesWritable,  BytesWritable, IntWritable, BytesWritable)

public synchronized void reduce(BytesWritable mer, Iterator<BytesWritable> values,Context context)
            throws IOException, InterruptedException {

有人有想法吗？

Answer 1

job.setInputFormatClass(SequenceFileInputFormat.class);

应该

job.setInputFormatClass(IntWritable.class);

您的映射器输入是 int 和 bytes，但在工作中您将两个输入都作为序列