hadoop深入研究:(十)——序列化与Writable接口

1.序列化的作用是什么？
2.hadoop在节点间的内部通讯使用的是什么？
3.hadoop自身的序列化存储格式有什么优势和劣势？

所有源码在github上，https://github.com/lastsweetop/styhadoop


简介
序列化和反序列化就是结构化对象和字节流之间的转换，主要用在内部进程的通讯和持久化存储方面。


通讯格式需求

hadoop在节点间的内部通讯使用的是RPC，RPC协议把消息翻译成二进制字节流发送到远程节点，远程节点再通过反序列化把二进制流转成原始的信息。RPC的序列化需要实现以下几点：

1.压缩，可以起到压缩的效果，占用的宽带资源要小。
2.快速，内部进程为分布式系统构建了高速链路，因此在序列化和反序列化间必须是快速的，不能让传输速度成为瓶颈。
3.可扩展的，新的服务端为新的客户端增加了一个参数，老客户端照样可以使用。
4.兼容性好，可以支持多个语言的客户端


存储格式需求

表面上看来序列化框架在持久化存储方面可能需要其他的一些特性，但事实上依然是那四点：
1.压缩，占用的空间更小
2.快速，可以快速读写
3.可扩展，可以以老格式读取老数据
4.兼容性好，可以支持多种语言的读写

hadoop的序列化格式

hadoop自身的序列化存储格式就是实现了Writable接口的类,他只实现了前面两点，压缩和快速。但是不容易扩展，也不跨语言。
我们先来看下Writable接口，Writable接口定义了两个方法：
1.将数据写入到二进制流中
2.从二进制数据流中读取数据

package org.apache.hadoop.io;  
  
public interface Writable {  
    void write(java.io.DataOutput p1) throws java.io.IOException;  
  
    void readFields(java.io.DataInput p1) throws java.io.IOException;  
}  
复制代码

我们再来看下Writable接口与序列化和反序列化是如何关联的：

package com.sweetop.styhadoop;  
  
import junit.framework.Assert;  
import org.apache.hadoop.io.IntWritable;  
import org.apache.hadoop.io.Writable;  
import org.apache.hadoop.util.StringUtils;  
import org.junit.Before;  
import org.junit.Test;  
  
import java.io.*;  
  
/** 
 * Created with IntelliJ IDEA. 
 * User: lastsweetop 
 * Date: 13-7-4 
 * Time: 下午10:25 
 * To change this template use File | Settings | File Templates. 
 */  
public class TestWritable {  
    byte[] bytes=null;  
  
    /** 
     * 初始化一个IntWritable实例，并且调用系列化方法 
     * @throws IOException 
     */  
    @Before  
    public void init() throws IOException {  
        IntWritable writable = new IntWritable(163);  
        bytes = serialize(writable);  
    }  
  
    /** 
     * 一个IntWritable序列号后的四个字节的字节流 
     * 并且使用big-endian的队列排列 
     * @throws IOException 
     */  
    @Test  
    public void testSerialize() throws IOException {  
        Assert.assertEquals(bytes.length,4);  
        Assert.assertEquals(StringUtils.byteToHexString(bytes),"000000a3");  
    }  
  
    /** 
     * 创建一个没有值的IntWritable对象，并且通过调用反序列化方法将bytes的数据读入到它里面 
     * 通过调用它的get方法，获得原始的值，163 
     */  
    @Test  
    public void testDeserialize() throws IOException {  
        IntWritable newWritable = new IntWritable();  
        deserialize(newWritable,bytes);  
        Assert.assertEquals(newWritable.get(),163);  
    }  
  
    /** 
     * 将一个实现了Writable接口的对象序列化成字节流 
     * @param writable 
     * @return 
     * @throws IOException 
     */  
    public static byte[] serialize(Writable writable) throws IOException {  
        ByteArrayOutputStream out = new ByteArrayOutputStream();  
        DataOutputStream dataOut = new DataOutputStream(out);  
        writable.write(dataOut);  
        dataOut.close();  
        return out.toByteArray();  
    }  
  
    /** 
     * 将字节流转化为实现了Writable接口的对象 
     * @param writable 
     * @param bytes 
     * @return 
     * @throws IOException 
     */  
    public static byte[] deserialize(Writable writable,byte[] bytes) throws IOException {  
        ByteArrayInputStream in=new ByteArrayInputStream(bytes);  
        DataInputStream dataIn = new DataInputStream(in);  
        writable.readFields(dataIn);  
        dataIn.close();  
        return bytes;  
    }  
}  
复制代码

WritableComparable和comparators
IntWritable实现了WritableComparable，接口看下源代码知道，WritableComparable是Writable接口和java.lang.Comparable<T>的一个子接口。

package org.apache.hadoop.io;  
  
public interface WritableComparable <T>  extends org.apache.hadoop.io.Writable, java.lang.Comparable<T> {  
}  
复制代码

MapReduce在排序部分要根据key值的大小进行排序，因此类型的比较相当重要，RawComparator是Comparator的增强版

package org.apache.hadoop.io;  
  
public interface RawComparator <T>  extends java.util.Comparator<T> {  
    int compare(byte[] bytes, int i, int i1, byte[] bytes1, int i2, int i3);  
}  
复制代码

它可以做到，不先反序列化就可以直接比较二进制字节流的大小:

package com.sweetop.styhadoop;  
  
import org.apache.hadoop.io.IntWritable;  
import org.apache.hadoop.io.RawComparator;  
import org.apache.hadoop.io.Writable;  
import org.apache.hadoop.io.WritableComparator;  
import org.eclipse.jdt.internal.core.Assert;  
import org.junit.Before;  
import org.junit.Test;  
  
import java.io.ByteArrayOutputStream;  
import java.io.DataOutputStream;  
import java.io.IOException;  
  
/** 
 * Created with IntelliJ IDEA. 
 * User: lastsweetop 
 * Date: 13-7-5 
 * Time: 上午1:26 
 * To change this template use File | Settings | File Templates. 
 */  
public class TestComparator {  
    RawComparator<IntWritable> comparator;  
    IntWritable w1;  
    IntWritable w2;  
  
    /** 
     * 获得IntWritable的comparator,并初始化两个IntWritable 
     */  
    @Before  
    public void init() {  
        comparator = WritableComparator.get(IntWritable.class);  
        w1 = new IntWritable(163);  
        w2 = new IntWritable(76);  
    }  
  
    /** 
     * 比较两个对象大小 
     */  
    @Test  
    public void testComparator() {  
        Assert.isTrue(comparator.compare(w1, w2) > 0);  
    }  
  
    /** 
     * 序列号后进行直接比较 
     * @throws IOException 
     */  
    @Test  
    public void testcompare() throws IOException {  
        byte[] b1 = serialize(w1);  
        byte[] b2 = serialize(w2);  
        Assert.isTrue(comparator.compare(b1, 0, b1.length, b2, 0, b2.length) > 0);  
    }  
  
    /** 
     * 将一个实现了Writable接口的对象序列化成字节流 
     * 
     * @param writable 
     * @return 
     * @throws java.io.IOException 
     */  
    public static byte[] serialize(Writable writable) throws IOException {  
        ByteArrayOutputStream out = new ByteArrayOutputStream();  
        DataOutputStream dataOut = new DataOutputStream(out);  
        writable.write(dataOut);  
        dataOut.close();  
        return out.toByteArray();  
    }  
}  
复制代码