Hadoop Serialize（二）

hadoop 序列化特性

在hadoop 权威指南中，有如此介绍hadoop序列化机制应有如下的特性。

Compact(紧凑)。紧凑的序列化机制可以充分利用数据中心的带宽。

Fast(快速)。进程间的通信时会大量使用序列化机制，减少序列化和反序列化的开销有利用性能的提升。

Extensible(可扩展)。系统升级，通信协议发生变化，类的定义发生变化，序列化机制需要支持这些升级和变化（用过java RMI的同学应该知晓如果服务端升级一个model类的定义是一件多么困难的事情）。

Interoperable(互操作)。可以支持不同开发语言间的通信。

java的序列化机制明显不符合上述要求。java Serialization将每个对象的类名写入输出流中，这导致了java序列化对象需要占用比原对象更多的存储空间。同时，为了减少数据量，同一个类的对象的序列化结果只输出一份元数据，并通过某种形式的引用，共享元数据。引用导致序列化后的流进行处理的时候，需要保持一些状态。但是对应mapreduce这种应用来说，上百G的文件，反序列化某个对象，需要访问文件中前面的缪一个元数据，这导致文件无法切割。同时，java序列化不断的创建新对象，对应mapreduce应用，不能重用对象，在已有对象上进行反序列化操作，而是不断创建反序列化的各种类型记录，这会带来大量的系统开销。

Writable

Hadoop引入了org.apache.hadoop.io.Writable接口。必须说明的是Writable依然不容易扩展到java以外的语言。

Writable接口有两个方法：

public interface Writable {
/**
* Serialize the fields of this object to <code>out</code>.
*
* @param out <code>DataOuput</code> to serialize this object into.
* @throws IOException
*/
void write(DataOutput out) throws IOException;

/**
* Deserialize the fields of this object from <code>in</code>.
*
* <p>For efficiency, implementations should attempt to re-use storage in the
* existing object where possible.</p>
*
* @param in <code>DataInput</code> to deseriablize this object from.
* @throws IOException
*/
void readFields(DataInput in) throws IOException;
}

自己实现一个writable，如下：

public class MyWritable implements Writable {

private int counter;
private long timestamp;

@Override
public void write(DataOutput out) throws IOException {
out.writeInt(counter);
out.writeLong(timestamp);
}

@Override
public void readFields(DataInput in) throws IOException {
counter = in.readInt();
timestamp = in.readLong();
}

public static MyWritable read(DataInput in) throws IOException {
MyWritable w = new MyWritable();
w.readFields(in);
return w;
}
}

在Hadoop的序列化机制中，还有几个接口比较重要。RawComparator和WritableComparable

public interface WritableComparable<T> extends Writable, Comparable<T> {

}

WritableComparable继承了Writable和Comparable接口。其中我们常接触的 IntWritable、ByteWritable、DoubleWritable等java基本类型对应的Writable类型都实现了WritableComparable接口。



RawComparator 接口允许执行这比较流中读取的未被反序列化的对象记录，从而省去了创建对象的所有开销。举例：

RawComparator<IntWritable> rawComparator = WritableComparator.get(IntWritable.class);

IntWritable w1 = new IntWritable(520);
IntWritable w2 = new IntWritable(521);

System.out.println(rawComparator.compare(w1,w2));

WritableComparator 实现了 RawComparator，充当了RawComparator实例的一个工厂方法。源码如下：

/** Get a comparator for a {@link WritableComparable} implementation. */
public static WritableComparator get(
Class<? extends WritableComparable> c, Configuration conf) {
WritableComparator comparator = comparators.get(c);
if (comparator == null) {
// force the static initializers to run
forceInit(c);
// look to see if it is defined now
comparator = comparators.get(c);
// if not, use the generic one
if (comparator == null) {
comparator = new WritableComparator(c, conf, true);
}
}
// Newly passed Configuration objects should be used.
ReflectionUtils.setConf(comparator, conf);
return comparator;
}