java - 在自定义 java 类中反序列化 MapWritable

Question

我目前正在尝试反序列化一个自定义对象，其中一个字段是 MapWritable，另一个字段是字符串。序列化似乎工作正常，但无法验证对象是否正在正确重新创建。他是我的字段和 write() readFields() 方法:

public class ExchangeDataSample implements DataSample {

    private String labelColumn;

    private MapWritable values = new MapWritable();

    ...other methods...

    @Override
    public void readFields(DataInput in) throws IOException {
        values.clear();
        values.readFields(in);
        labelColumn = in.readLine();
    }

    @Override
    public void write(DataOutput out) throws IOException {
        values.write(out);
        out.writeBytes(labelColumn);
    }
}

我在 MapReduce 作业中不断收到此异常:

java.lang.Exception: java.io.EOFException
    at org.apache.hadoop.mapred.LocalJobRunner$Job.runTasks(LocalJobRunner.java:462)
    at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:522)
Caused by: java.io.EOFException
    at java.io.DataInputStream.readFully(DataInputStream.java:197)
    at java.io.DataInputStream.readUTF(DataInputStream.java:609)
    at java.io.DataInputStream.readUTF(DataInputStream.java:564)
    at org.apache.hadoop.io.AbstractMapWritable.readFields(AbstractMapWritable.java:207)
    at org.apache.hadoop.io.MapWritable.readFields(MapWritable.java:167)
    at decisiontree.data.ExchangeDataSample.readFields(ExchangeDataSample.java:98)
    at org.apache.hadoop.io.ArrayWritable.readFields(ArrayWritable.java:96)
    at org.apache.hadoop.io.serializer.WritableSerialization$WritableDeserializer.deserialize(WritableSerialization.java:71)
    at org.apache.hadoop.io.serializer.WritableSerialization$WritableDeserializer.deserialize(WritableSerialization.java:42)
    at org.apache.hadoop.mapreduce.task.ReduceContextImpl.nextKeyValue(ReduceContextImpl.java:146)
    at org.apache.hadoop.mapreduce.task.ReduceContextImpl.nextKey(ReduceContextImpl.java:121)
    at org.apache.hadoop.mapreduce.lib.reduce.WrappedReducer$Context.nextKey(WrappedReducer.java:302)
    at org.apache.hadoop.mapreduce.Reducer.run(Reducer.java:170)
    at org.apache.hadoop.mapred.Task$NewCombinerRunner.combine(Task.java:1688)
    at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.sortAndSpill(MapTask.java:1637)
    at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.flush(MapTask.java:1489)
    at org.apache.hadoop.mapred.MapTask$NewOutputCollector.close(MapTask.java:723)
    at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:793)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:341)
    at org.apache.hadoop.mapred.LocalJobRunner$Job$MapTaskRunnable.run(LocalJobRunner.java:243)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

我非常感谢您的帮助。谢谢。

Answer 1

您收到此异常是因为您在读取时没有检查文件末尾。尝试将 readFields 方法更改为:

 @Override
public void readFields(DataInput in) throws IOException {
    values.clear();
    byte[] b = new byte[1024];
    //checks for end of file
    if(((DataInputStream) in).read(b)!=-1){ 
      values.readFields(in);
      labelColumn = in.readLine();
    }
}