Hello all, I am having a problem writing my own RecordReader. The basic setup I have is a large byte array that needs to be diced up into key value pairs such that the key is the index into the array and the value is a byte array itself (albeit much smaller). Here is the code that I currently have written to do just this:
/* This method is just the constructor for my new RecordReader public TrainingRecordReader(Configuration job, FileSplit split) throws IOException { start = split.getStart(); end = start + split.getLength(); final Path file = split.getPath(); compressionCodecs = new CompressionCodecFactory(job); final CompressionCodec codec = compressionCodecs.getCodec(file); // open the file and seek to the start of the split FileSystem fs = file.getFileSystem(job); FSDataInputStream fileIn = fs.open(split.getPath()); in = new TrainingReader(fileIn, job); this.pos = start; } // This returns the key, value pair I was talking about public synchronized boolean next(LongWritable key, BytesWritable value) throws IOException { if (pos >= end) return false; key.set(pos); // key is position int newSize = in.readVector(value); if (newSize > 0) { pos += newSize; return true; } return false; } // This extracts that smaller byte array from the large input file public int readVector(BytesWritable value) throws IOException { int numBytes = in.read(buffer); value.set(buffer, 0, numBytes); return numBytes; } So all of this worked just fine when I set conf.set("mapred.job.tracker", "local"), but now that I am attempting to test in a fake distributed setting (aka still one node, but I haven't set the above config param), I do not get what I want. Instead of getting unique key value pairs, I get repeated key value pairs based on the number of map tasks I have set. So, say that my large file contained 49 entries, I would want a unique key value pair for each of those, but if I set my numMapTasks to 7, I get 7 unique ones that repeat every 7 key value pairs. So it seems that each MapTask which ultimately calls my TrainingReader.next() method from above is somehow pointing to the same FileSplit. I know that in the LineRecordReader in the source there is some small little routine that skips the first line of the data if you aren't at the beginning???? Is that related? Why isn't it the case that split.getStart() isn't returning the absolute pointer to the start of the split? So many questions I don't know the answer to, haha. I would appreciate anyone's help in resolving this issue. Thanks very much! Cheers, Sean M. Arietta -- View this message in context: http://www.nabble.com/RecordReader-Functionality-tp18199187p18199187.html Sent from the Hadoop core-user mailing list archive at Nabble.com.