java.lang.RuntimeException: java.io.EOFException at org.apache.hadoop.io.WritableComparator.compare(WritableComparator.java:103)

[Tali K]

HI All,

I am getting this Exception on a cluster(10 nodes)  when I am running  simple 
hadoop map / reduce job.
I don't have this Exception while running it on my desktop in hadoop's pseudo 
distributed mode.
Can somebody help? I would really appreciate it.


10/09/29 14:28:34 INFO mapred.JobClient:  map 100% reduce 30%
10/09/29 14:28:36 INFO mapred.JobClient: Task Id : 
attempt_201009291306_0004_r_000000_0, Status : FAILED
java.lang.RuntimeException: java.io.EOFException
        at 
org.apache.hadoop.io.WritableComparator.compare(WritableComparator.java:103)
        at org.apache.hadoop.mapred.Merger$MergeQueue.lessThan(Merger.java:373)
        at org.apache.hadoop.util.PriorityQueue.upHeap(PriorityQueue.java:123)
        at org.apache.hadoop.util.PriorityQueue.put(PriorityQueue.java:50)
        at org.apache.hadoop.mapred.Merger$MergeQueue.merge(Merger.java:447)
        at org.apache.hadoop.mapred.Merger$MergeQueue.merge(Merger.java:381)
        at org.apache.hadoop.mapred.Merger.merge(Merger.java:107)
        at org.apache.hadoop.mapred.Merger.merge(Merger.java:93)
        at 
org.apache.hadoop.mapred.ReduceTask$ReduceCopier.createKVIterator(ReduceTask.java:2316)
        at 
org.apache.hadoop.mapred.ReduceTask$ReduceCopier.access$400(ReduceTask.java:576)
        at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:389)
        at org.apache.hadoop.mapred.Child.main(Child.java:170)
Caused by: java.io.EOFException
        at java.io.DataInputStream.readInt(DataInputStream.java:375)
        at speeditup.MsRead.readFields(MsRead.java:84)
        at 
org.apache.hadoop.io.WritableComparator.compare(WritableComparator.java:97)
        ... 11 more
Here is a class that has WritableComparator.compare. It has only 2 strings max 
length 20 characters for each.

public class MsRead implements WritableComparable < MsRead> {
    private static final Log LOG = 
LogFactory.getLog(speeditup.CalculateMinEvalue.class);
    
    private String query_id;
    
    private String record;
    
    
    
    public String getRecord() {
        return record;
    }
    public void setRecord(String record) {
        this.record = record;
    }
    
    public String  getQuery_id() {
        return query_id;
    }
    
    public void setQuery_id(String queryId) {
        query_id = queryId;
    }
    
    public MsRead()
    {
        ;
    }
    public MsRead(String a,  String r) 
    {
        setQuery_id(a);
        
        setRecord(r);
    }
    
    
    
    
      
    @Override
        public void readFields(DataInput in) throws IOException {
         LOG.debug("******myreadFields" + " "  );    
         LOG.warn("******myreadFields" + " "  );
         LOG.info("******myreadFields" + " " );
          query_id = in.readUTF();
          record = in.readUTF();
          
    }
    @Override
     public void write(DataOutput out) throws IOException {
          out.writeUTF(query_id);
          out.writeUTF(record);
          
     }
    
    
    

    
    public static class FirstComparator extends WritableComparator {
        
        private static final Text.Comparator TEXT_COMPARATOR = new 
Text.Comparator();
        
        public FirstComparator() {
          super(MsRead.class);
        }

        @Override
        public int compare(byte[] b1, int s1, int l1,
                           byte[] b2, int s2, int l2) {
         
          try {
            int firstL1 = WritableUtils.decodeVIntSize(b1[s1]) + readVInt(b1, 
s1);
            int firstL2 = WritableUtils.decodeVIntSize(b2[s2]) + readVInt(b2, 
s2);
            return TEXT_COMPARATOR.compare(b1, s1, firstL1, b2, s2, firstL2);
          } catch (IOException e) {
            throw new IllegalArgumentException(e);
          }
            
        }
        
        @Override
        public int compare(WritableComparable a, WritableComparable b) {
          if (a instanceof MsRead && b instanceof MsRead) {
         
            //System.err.println("COMPARE " + ((MsRead)a).getType() + "\t" + 
((MsRead)b).getType() + "\t" 
                //    + (((MsRead) a).toString().compareTo(((MsRead) 
b).toString()))); 
            return (((MsRead) a).toString().compareTo(((MsRead) b).toString()));
           
          }
          return super.compare(a, b);
        }
        
        
      }
    
        @Override
        public int compareTo(MsRead o) {
         return this.toString().compareTo(o.toString());
      }
        @Override
        public boolean equals(Object right) {
            if (right instanceof MsRead )
            {
                return (query_id.equals(((MsRead)right).query_id));
            }
            else 
                return false;
        }
        @Override
        public int hashCode() {
            return query_id.hashCode() ;
        }
       
        @Override 
        public String toString()
        {
            return query_id;
        }
        public String toOutputString()
        {
            return record;
        }
        
}