You are right, there is no readInt , I have only 2 String fields in MsRead
Here are lines: I'll also will send both files in attch.
Thanks in advance for your help.
@Override
public void readFields(DataInput in) throws IOException {
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() ;
}
> Date: Wed, 29 Sep 2010 22:27:15 -0700
> Subject: Re: java.lang.RuntimeException: java.io.EOFException at
> org.apache.hadoop.io.WritableComparator.compare(WritableComparator.java:103)
> From: [email protected]
> To: [email protected]
>
> Your MsRead.readFields() doesn't contain readInt().
> Can you show us the lines around line 84 of MsRead.java ?
>
> On Wed, Sep 29, 2010 at 2:44 PM, Tali K <[email protected]> wrote:
>
> >
> > 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;
> > }
> >
> > }
> >
package speeditup;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordReader;
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.util.GenericOptionsParser;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.Random;
import java.util.StringTokenizer;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.hadoop.util.*;
public class CalculateMinEvalue {
private static final Log LOG = LogFactory.getLog(speeditup.CalculateMinEvalue.class);
//static Logger log = Logger.getLogger("UniqueQueryIdCount");
public static class Map extends Mapper< LongWritable, Text, MsRead, DoubleWritable> {
private final static DoubleWritable one = new DoubleWritable(1);
public void map(LongWritable key, Text value, Context output) throws IOException, InterruptedException {
String line = value.toString();
if (line.isEmpty())
return;
StringTokenizer tokenizer = new StringTokenizer(line);
if (tokenizer.countTokens() != 12 ) return;
String query_id = tokenizer.nextToken();
String subject_id = tokenizer.nextToken();
if (query_id.equals(subject_id))
return;
for (int i = 0; i < 8; i++)
tokenizer.nextToken();
String e_value = tokenizer.nextToken();
MsRead read = new MsRead(query_id, line);
double eval = new Double(e_value).doubleValue();
//System.err.println(read.getRecord());
one.set(eval);
output.write(read, one);
}// end method
}//end class
public static class Combine extends Reducer<MsRead, DoubleWritable, MsRead, DoubleWritable>
{
public void reduce(MsRead key, Iterable<DoubleWritable> values, Context output
) throws IOException, InterruptedException
{ double min = Double.MAX_VALUE;
for (DoubleWritable val : values) {
min = Math.min(min,val.get());
}
output.write(key, new DoubleWritable(min));
//report.progress();
}
}
//=====================================================================================================
public static class Reduce extends Reducer<MsRead, DoubleWritable, String, DoubleWritable>
{
Text textw = new Text();
public void reduce(MsRead key, Iterable<DoubleWritable> values, Context output
) throws IOException, InterruptedException
{ double min = Double.MAX_VALUE;
for (DoubleWritable val : values) {
min = Math.min(min,val.get());
}
output.write(key.getRecord(), new DoubleWritable(min));
//report.progress();
}
}
// static String []args1 = {"input", "output1"} ;
static public void main (String [] args) throws Exception
{
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length != 2) {
System.err.println("Usage: calculate_min_value <in> <out>");
System.exit(2);
}
// UniqueQueryIdCount.class);
Job job = new Job(conf, "msread_processor");
job.setJarByClass(CalculateMinEvalue.class);
// the map output is MsPair, IntWritable
job.setMapOutputKeyClass(MsRead.class);
job.setMapOutputValueClass(DoubleWritable.class);
// the reducer
//job.setInputFormatClass(RecordReader.class);
job.setMapperClass(Map.class);
job.setCombinerClass(Combine.class);
job.setReducerClass(Reduce.class);
job.setOutputKeyClass(String.class);
job.setOutputValueClass(DoubleWritable.class);
FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1] + Integer.toString(new Random().nextInt(Integer.MAX_VALUE))));
System.err.println("Usage: calculate_min_value <in> <out> STARTED");
LOG.warn("******DONE" + " " + new Date());
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
package speeditup;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
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.LongWritable;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapreduce.RecordReader;
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.util.GenericOptionsParser;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.hadoop.util.*;
public class MsRead implements WritableComparable < MsRead> {
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 MsRead()
{
;
}
public MsRead(String a, String t)
{
setQuery_id((a));
setRecord(t);
}
public void setQuery_id(String queryId) {
query_id = queryId;
}
@Override
public void readFields(DataInput in) throws IOException {
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;
}
}
