这是我的代码:

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
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.io.WritableComparable;
import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class SecondarySort extends Configured implements Tool{

public static void main(String[] args) {
    try {
        ToolRunner.run(new Configuration(), new SecondarySort(), args);
    } catch (Exception e) {
        e.printStackTrace();
    }
}

static class KeyPartitioner implements Partitioner<StockKey, DoubleWritable> {

    @Override
    public int getPartition(StockKey arg0, DoubleWritable arg1, int arg2) {

        int partition = arg0.name.hashCode() % arg2;

        return partition;
    }

    @Override
    public void configure(JobConf job) {
    }

}


static class StockKey implements WritableComparable<StockKey> {

    String name;
    Long timestamp;

    public StockKey() {

    }

    StockKey(String name, Long timestamp){
        this.name = name;
        this.timestamp = timestamp;
    }

    @Override
    public void readFields(DataInput arg0) throws IOException {
        name = WritableUtils.readString(arg0);
        timestamp = arg0.readLong();
    }

    @Override
    public void write(DataOutput arg0) throws IOException {
        WritableUtils.writeString(arg0, name);
        arg0.writeLong(timestamp);
    }

    @Override
    public int compareTo(StockKey arg0) {
        int result = 0;

        result = name.compareToIgnoreCase(arg0.name);

        if(result == 0)
            result = timestamp.compareTo(arg0.timestamp);   

        return result;
    }

    public String toString() {
        String outputString = name+","+timestamp;
        return outputString;
    }
}


static class StockReducer implements Reducer<StockKey, DoubleWritable, Text, Text>{

    public void reduce(StockKey key, Iterator<DoubleWritable> value, Outp      
OutputCollector<Text, Text> context, Reporter reporter) 
                throws IOException {

        Text k = new Text(key.toString());

        while(value.hasNext()) {

            Double v = value.next().get();
            Text t = new Text(v.toString());

            context.collect(k, t);          
        }
    }

    @Override
    public void configure(JobConf job) {
        // TODO Auto-generated method stub

    }

    @Override
    public void close() throws IOException {
        // TODO Auto-generated method stub

    }
}


static class StockMapper implements Mapper<LongWritable, Text, StockKey, 
DoubleWritable> {

    public void map(LongWritable offset, Text value, OutputCollector<StockKey, 
DoubleWritable> context, Reporter reporter) 
                throws IOException {

        String[] values = value.toString().split(",");
        StockKey key = new StockKey(values[0].trim(), 
Long.parseLong(values[1].trim()));
        DoubleWritable val = new 
DoubleWritable(Double.parseDouble(values[2].trim()));


            context.collect(key, val);

    }

    @Override
    public void configure(JobConf job) {
        // TODO Auto-generated method stub

    }

    @Override
    public void close() throws IOException {
        // TODO Auto-generated method stub

    }

}

@SuppressWarnings("unchecked")
@Override
public int run(String[] arg) throws Exception {

    JobConf conf = new JobConf(getConf(), SecondarySort.class);
    conf.setJobName(SecondarySort.class.getName());

    conf.setJarByClass(SecondarySort.class);

    conf.setInputFormat(TextInputFormat.class);
    conf.setOutputFormat(TextOutputFormat.class);

    conf.setMapOutputKeyClass(StockKey.class);
    conf.setMapOutputValueClass(Text.class);

    conf.setPartitionerClass((Class<? extends Partitioner<StockKey, 
DoubleWritable>>) KeyPartitioner.class);

    conf.setMapperClass((Class<? extends Mapper<LongWritable, Text, StockKey, 
DoubleWritable>>) StockMapper.class);
    conf.setReducerClass((Class<? extends Reducer<StockKey, DoubleWritable, 
Text, Text>>) StockReducer.class);

    FileInputFormat.addInputPath(conf, new Path(arg[0]));
    FileOutputFormat.setOutputPath(conf, new Path(arg[1]));

    JobClient.runJob(conf);

    return 0;
}

}

异常(exception)情况:

java.io.IOException: Type mismatch in value from map: expected 
org.apache.hadoop.io.Text, recieved org.apache.hadoop.io.DoubleWritable
    at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.collect(MapTask.java:876)
    at org.apache.hadoop.mapred.MapTask$OldOutputCollector.collect(MapTask.java:499)
    at SecondarySort$StockMapper.map(SecondarySort.java:135)
    at SecondarySort$StockMapper.map(SecondarySort.java:1)
    at org.apache.hadoop.mapred.MapRunner.run(MapRunner.java:50)
    at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:391)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:325)
    at org.apache.hadoop.mapred.Child$4.run(Child.java:270)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:396)
    at     
org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1127)
    at org.apache.hadoop.mapred.Child.main(Child.java:264)


12/07/13 03:22:32 INFO mapred.JobClient: Task Id :   
attempt_201207130314_0002_m_000001_2, Status : FAILED

java.io.IOException: Type mismatch in value from map: expected 
org.apache.hadoop.io.Text, recieved org.apache.hadoop.io.DoubleWritable
    at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.collect(MapTask.java:876)
    at org.apache.hadoop.mapred.MapTask$OldOutputCollector.collect(MapTask.java:499)
    at SecondarySort$StockMapper.map(SecondarySort.java:135)
    at SecondarySort$StockMapper.map(SecondarySort.java:1)
    at org.apache.hadoop.mapred.MapRunner.run(MapRunner.java:50)
    at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:391)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:325)
    at org.apache.hadoop.mapred.Child$4.run(Child.java:270)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:396)
    at 
org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1127)
    at org.apache.hadoop.mapred.Child.main(Child.java:264)

最佳答案

这段代码有很多潜在的问题可能会导致它:

• StockKey - 你应该覆盖默认的 hashCode() 方法 - 目前两个 StockKey 具有相同的内容将具有不同的 hashCode 值(就好像您不覆盖 JVM 默认值一样，那么它将返回一个数字，该数字在所有范围内都是两个对象在内存中的地址)。我知道在您的分区程序中您只使用 name 字段(它是一个字符串并且将具有 hashCode() 的有效实现，但这是一个很好的做法，以防将来您使用整个 Stock 对象的 hashCode() 并想知道为什么两个相同的 Stock 对象最终出现在不同的 reducer 中

• KeyPartitioner - 您需要 Math.abs(..) arg0.name.hashCode() 的结果.目前，这个值可能会返回负数，当你对 reducer 的数量取模时，将返回一个负数。链式效应是 MR 框架将抛出​​异常，因为它期望一个介于 0(含)和 reducer 数量(不含)之间的数字。这可能是你的问题所在，我将在下一点解释

• Mapper.map 方法 - 当您调用 context.collect 时，您正在吞下任何潜在的输出异常。继续我之前关于分区程序的观点——如果它返回一个负数，将抛出一个异常，你需要处理这个异常。在某些情况下捕获和吞噬异常可能没问题(例如输入记录的数据验证)，但是输出时发生的任何异常都应该被抛给 MR 框架以标记出现问题并且此映射器的输出是错误的/不完整:

  try {
      context.collect(key, val);
  } catch (IOException e) {
      e.printStackTrace();
  }
  

• 最后，您需要显式声明您的 map 和 reduce 输出类型(这会导致异常，因为您当前将 map 值输出类型声明为 Text，而实际上 mapper 输出的是 DoubleWritable):

job.setMapOutputKeyClass(StockKey.class); job.setMapOutputValueClass(DoubleWritable.class);

job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class);

我建议您删除 context.collect 调用周围的 try/catch block 并重新运行您的作业(或者只检查 map task 的日志，看看您是否看到堆栈跟踪)。

关于java - Mapreduce作业运行，出现异常，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/11447736/