Hadoop-2.4.1学习之Writable及其实现

      Hadoop基于DataInput和DataOutput实现了简单、高效的序列化协议，而Writable接口定义了Hadoop序列化的方法，MapReduce框架中的任何键值类型都实现了该接口，比如IntWritable、LongWritable等，详细的类关系见下图：


      通过上图可以发现，MapReduce中常用的键值类型都没有直接实现Writable接口，而是实现了该接口的子接口WritableComparable，该子接口还继承了Comparable接口，这意味着实现类除了可以序列化和发序列化外，还能够彼此进行比较，这是因为当这些类做为键在MapReduce中使用时，在排序阶段需要对其进行比较排序。但这并不是说如果实现自定义的序列化类时，必须实现WritableComparable接口，仅当自定义的序列化类也用作键时才必须实现该接口，如果仅是做为值使用的话，仅实现Writable接口即可。
      当自定义的序列化类用做键时，需要考虑到在根据键进行分区时经常用到hashCode()方法，因此需要确保该方法在不同的JVM实例中返回相同的结果，而Object对象中默认的hashCode()方法不能够满足该特性，所以在实现自定义类时需要重写hashCode()方法，而如果两个对象根据equals()方法是相等的，那么二者的hashCode()返回值也必须相同，因此在重写hashCode()的时候，有必要重写equals(Object obj)方法。
      除了上图中实现WritableComparable的类外，还有若干类直接实现了Writable接口，比如ObjectWritable，一个多态的Writable，该类不使用Writable封装就能够处理数组、字符串和其它Java基本类型。还有Writable集合类：ArrayWritable、EnumSetWritable、MapWritable、TwoDArrayWritable、SortedMapWritable。其中ArrayWritable是对相同Writable类型的数组的封装，也就是该类中Writable的类型必须相同，是IntWritable都是IntWritable，不能既有IntWritable也有LongWritable。TwoDArrayWritable是对二维数组即矩阵的封装，同样该类中Writable的类型也必须相同。EnumSetWritable是对EnumSet封装的Writable，MapWritable实现了Map<Writable,Writable>接口，SortedMapWritable实现了SortedMap<WritableComparable,Writable>接口，二者当然也都实现了Writable接口，在二者的内部实现中，使用Byte类型指示指定的类型，因此在一个Map实例中最多只能有127个不同的类：/* Class to id mappings */
@VisibleForTesting
Map<Class, Byte> classToIdMap = new ConcurrentHashMap<Class, Byte>();
/* Id to Class mappings */
@VisibleForTesting
Map<Byte, Class> idToClassMap = new ConcurrentHashMap<Byte, Class>();
      现在通过分析IntWritable和Text的源代码来学习如何编写Writable以及WritableComparable，首先是IntWritable的源代码：public class IntWritable implements WritableComparable<IntWritable> {
private int value;
public IntWritable() {}
public IntWritable(int value) { set(value); }
/** Set the value of this IntWritable. */
public void set(int value) { this.value = value; }
/** Return the value of this IntWritable. */
public int get() { return value; }
@Override
//重写Writable中的readFields(DataInput in)
public void readFields(DataInput in) throws IOException {
value = in.readInt();
}
@Override
//重写Writable中的write(DataOutput out)
public void write(DataOutput out) throws IOException {
out.writeInt(value);
}
/** Returns true if <code>o</code> is a IntWritable with the same value. */
@Override
public boolean equals(Object o) {
if (!(o instanceof IntWritable))
return false;
IntWritable other = (IntWritable)o;
return this.value == other.value;
}
@Override
public int hashCode() {
return value;
}
/** Compares two IntWritables. */
@Override
//重写Comparable接口中的compareTo方法
public int compareTo(IntWritable o) {
int thisValue = this.value;
int thatValue = o.value;
return (thisValue<thatValue ? -1 : (thisValue==thatValue ? 0 : 1));
}
@Override
public String toString() {
return Integer.toString(value);
}
//此处省略了继承自WritableComparator的内部类Comparator
static { // register this comparator
WritableComparator.define(IntWritable.class, new Comparator());
}
}
      IntWritable的源码相对来说还是比较简单的，除了实现接口中的方法外，还重写了hashCode、equals和toString方法，这也是要注意的一点。其次是Text类，Text将字符串存储为标准UTF8编码，提供了在字节层次序列化、反序列化和比较字符串的方法，比如decode(byte[]utf8)、encode(String string)、readFields(DataInput in)、write(DataOutput out)等。该类除了实现WritableComparable外，还继承自BinaryComparable抽象类，其中实现的方法如下：private byte[] bytes;
private int length;
@Override
public void readFields(DataInput in) throws IOException {
//从输入流中读取整数值，更多工具方法可参考WritableUtils工具类
int newLength = WritableUtils.readVInt(in);
setCapacity(newLength, false);
//向bytes中读入长度为newLength的数据
in.readFully(bytes, 0, newLength);
length = newLength;
}
@Override
public void write(DataOutput out) throws IOException {
WritableUtils.writeVInt(out, length);
out.write(bytes, 0, length);
}
@Override
public int compareTo(BinaryComparable other) {
if (this == other)
return 0;
return WritableComparator.compareBytes(getBytes(), 0, getLength(),
other.getBytes(), 0, other.getLength());
}
      总结IntWritable和Text类的实现，可以据此实现自定的WritableComparable，下面就是一个简单的示例。在该示例中使用name和age做为联合键，只有在二者都相同的情况下才认为是一个对象。public class CompositeWritable implements WritableComparable<CompositeWritable>{
private String name;
private int age;
public CompositeWritable(){}
public CompositeWritable(String name, int age){
set(name, age);
}
@Override
public void readFields(DataInput in) throws IOException {
name = in.readUTF();
age = in.readInt();
}
@Override
public void write(DataOutput out) throws IOException {
out.writeUTF(name);
out.writeInt(age);
}
@Override
public int compareTo(CompositeWritable o) {
int cmp = name.compareTo(o.getName());
if(cmp != 0)
return cmp;
return age < o.getAge()? -1:(age == o.getAge()? 0 : 1);
}
@Override
public boolean equals(Object o) {
if(o instanceof CompositeWritable){
CompositeWritable other = (CompositeWritable)o;
return this.name.equals(other.name) && this.age == other.age;
}
return false;
}
@Override
public int hashCode() {
return name.hashCode() + age;
}
@Override
public String toString() {
return name + "\t" + age;
}
public void set(String name, int age){
this.name = name;
this.age = age;
}
public String getName(){
return this.name;
}
public int getAge(){
return this.age;
}
}