Thanks Owen for your reply !
The terasort input you have implemented is text type. And the input is line
format where as I am dealing with sequence binary file. For my requirement I
have created two writable implementables for the key and value respectively
:
*FpMetaId : key*
public class FpMetaId extends BytesWritable {
public long fp;
*// 8 bytes key*
public FpMetaId (long fp)
{
this.fp = fp;
}
public FpMetaId ()
{
this (0);
}
public void readFields(DataInput in) throws IOException {
setSize(0); // clear the old data
//setSize(in.readInt());
setSize(8);
in.readFully(getBytes(), 0, getSize());
fp = this.byteArrayToLong (getBytes(), 0);
}
public void write(DataOutput out) throws IOException {
//out.writeInt(size);
setSize(8);
byte[] bytes = this.longToByteArray (fp);
out.write(getBytes(), 0, getSize());
}
public static byte[] intToByteArray(int value)
{
byte[] b = new byte[4];
for (int i = 0; i < 4; i++)
{
int offset = (b.length - 1 - i) * 8;
b[i] = (byte) ((value >>> offset) & 0xFF);
}
return b;
}
public static byte[] longToByteArray(long value)
{
byte[] b = new byte[8];
for (int i = 0; i < 8; i++) {
int offset = (b.length - 1 - i) * 8;
b[i] = (byte) ((value >> offset) & 0xFF);
}
return b;
}
public static final int byteArrayToInt(byte [] b, int offset) {
return (b[offset + 0] << 24)
+ ((b[offset + 1] & 0xFF) << 16)
+ ((b[offset + 2] & 0xFF) << 8)
+ (b[offset + 3] & 0xFF);
}
public static final long byteArrayToLong(byte [] b, int offset) {
return (b[offset + 0] << 24)
+ ((b[offset + 1] & 0xFF) << 16)
+ ((b[offset + 2] & 0xFF) << 8)
+ (b[offset + 3] & 0xFF);
}
}
********************************************************************************************
*FpMetadata ---> value *
*//32 bytes value*
public class FpMetadata extends BytesWritable {
public long fbn;
public int ino;
public int wi_gen;
public int cp_count;
public int unprocessed;
public int compress_attempted;
public int gatherer;
public FpMetadata (long fbn, int ino, int wi_gen, int cp_count,int
unprocessed, int compress_attempted, int gatherer)
{
super();
this.fbn = fbn;
this.ino = ino;
this.wi_gen = wi_gen;
this.cp_count = cp_count;
this.unprocessed = unprocessed;
this.compress_attempted = compress_attempted;
this.gatherer = gatherer;
}
public FpMetadata ()
{
this (0,0,0,0,1,1,1);
}
public void readFields(DataInput in) throws IOException {
setSize(0); // clear the old data
setSize(32);
in.readFully(getBytes(), 0,getSize());
fbn = this.byteArrayToLong (getBytes(), 0);
ino = this.byteArrayToInt (getBytes(), 8);
wi_gen = this.byteArrayToInt (getBytes(), 12);
cp_count = this.byteArrayToInt (getBytes(), 16);
unprocessed = this.byteArrayToInt (getBytes(), 20);
compress_attempted = this.byteArrayToInt (getBytes(), 24);
gatherer = this.byteArrayToInt (getBytes(), 28);
}
public void write(DataOutput out) throws IOException {
//out.writeInt(size);
setSize(32);
byte[] bytes = this.longToByteArray (fbn);
bytes = concat(bytes, intToByteArray (ino));
bytes = concat(bytes, intToByteArray (wi_gen));
bytes = concat(bytes, intToByteArray (cp_count));
bytes = concat(bytes, intToByteArray (unprocessed));
bytes = concat(bytes, intToByteArray (compress_attempted));
bytes = concat(bytes, intToByteArray (gatherer));
out.write(getBytes(), 0,getSize());
}
public static byte[] intToByteArray(int value) {
byte[] b = new byte[4];
for (int i = 0; i < 4; i++) {
int offset = (b.length - 1 - i) * 8;
b[i] = (byte) ((value >> offset) & 0xFF);
}
return b;
}
public static byte[] longToByteArray(long value)
{
byte[] b = new byte[8];
for (int i = 0; i < 8; i++) {
int offset = (b.length - 1 - i) * 8;
b[i] = (byte) ((value >> offset) & 0xFF);
}
return b;
}
public static final int byteArrayToInt(byte [] b, int offset) {
return (b[offset + 0] << 24)
+ ((b[offset + 1] & 0xFF) << 16)
+ ((b[offset + 2] & 0xFF) << 8)
+ (b[offset + 3] & 0xFF);
}
public static final long byteArrayToLong(byte [] b, int offset) {
return (b[offset + 0] << 24)
+ ((b[offset + 1] & 0xFF) << 16)
+ ((b[offset + 2] & 0xFF) << 8)
+ (b[offset + 3] & 0xFF);
}
public static byte[] concat(byte[] a, byte[] b)
{
byte[] result = new byte[a.length + b.length];
System.arraycopy(a, 0, result, 0, a.length);
System.arraycopy(b, 0, result, a.length, b.length);
return result;
}
}
*******************************************************************888
I assume I should also implement a inputformat and outputformat along with
these. But I am not able to figure out how to provide the respective
filesplit and recordreader/writer.
Also SequenceFile which is apt for binary sequence files has a record
structure as <record len><key len><key><value> and this is what is
implemented by all sequence supporting recordreader. In that case how can I
use any of these recordreaders since my records are in the format :
<key><value>...
Please reply..
Thanks
Matthew
