hadoop 023.0与hadoop 1.0 io.serializable分析

原创
2012/01/01 16:06
阅读数 447

   hadoop升级到0.23.0和1.0版本后,其IO底层除了自己实现的Writable序列化后,还增加了一个io.serializer包,该包提供了一种可插拔的持久化框架(Pluggable Serialization Framework)。之所以说是可插拔的,是因为可以把现存的持久化方式嵌入到hadoop的程序中。这个包中有以下的一些类: 

    接口类:

          Serializer:定义了序列化的接口

       Serialization:定义了序列化的机制(作为序列化接口Serializer和反序列化接口Deserializer的上层抽象,提供了一个抽象的序列化和反序列化实例的返回)

Deserializer :定义了反序列化的接口

Serializer接口(具体的序列化类实现接口)

Serializer接口位于hadoop.io. Serializer包下面,为hadoop的序列化提供了一种机制。(This package provides a mechanism for using different serialization frameworks in Hadoop

该接口包含了三个方法:

    void open(OutputStream out)

       --打开一个输出流为序列化做准备。

void serialize(T t)

    --对一个对象序列化

void close()

    --关闭输出流

 

 

Serialization接口(序列化框架接口)

该接口作为序列化和反序列化内部封装的一个部分,提供了下面三个方法。

accept(Class<?> c)

    --查看参数Class类是否支持序列化。

Deserializer<T>   getDeserializer(Class<T> c)

    --获取反序列化对象

Serializer<T> getSerializer(Class<T> c)

    --获取序列化对象

 

Deserializer接口

该接口提供了反序列化的机制,与Serializer接口相对应,该接口同样提供了三个方法。

<< 

Provides a facility for deserializing objects of type from an InputStream.

Deserializers are stateful, but must not buffer the input since other producers may read from the input between calls to deserialize(Object).

>> 

void   close()

    --关闭输入流

T   deserialize(T t)

    --对一个对象反序列化

    Deserialize(T t)方法的实现:

    @SuppressWarnings("unchecked")

    public T deserialize(T object) throws IOException {

      try {

        // ignore passed-in object

        return (T) ois.readObject();

      } catch (ClassNotFoundException e) {

        throw new IOException(e.toString());

      }

    }

void   open(InputStream in)

    --打开一个输入流实现反序列化做准备

    Open方法的实现为:

    private ObjectInputStream ois;

 

    public void open(InputStream in) throws IOException {

      ois = new ObjectInputStream(in) {

        @Override protected void readStreamHeader() {

          // no header

        }

      };

}

 

注意:open方法利用装饰器模式为InputStream装饰为ObjectInputStream

      Deserialize方法从open方法装饰后的ObjectInputStream类中读取反序列化的对象,从源码中可以看出Deserialize并未对ObjectInputStream进行验证,这也就意味着必须先open->Deserialize否则会抛出异常。

其他的类:

 JavaSerilization:实现Serialization接口,并维护两个分别实现SerializerDeserializer接口的内部类。这两个内部类包装Java的序列化机制,实现对实现Serializable接口的类的序列化。 

说明:

    1.open方法利用装饰器模式为InputStream装饰为ObjectInputStream

    2. Deserialize方法从open方法装饰后的ObjectInputStream类中读取反序列化的对象,从源码中可以看出Deserialize并未对ObjectInputStream进行验证,这也就意味着必须先open->Deserialize否则会抛出异常。

结构如下:

    

  WritableSerialization:同上,实现对实现Writable接口的类的序列化 

说明:

1.open方法利用装饰器模式为InputStream装饰为DataInputStream 

2. Deserialize方法实际上调用的是Writable类的writeDataOutPut)方法和Writable.readFields(java.io.DataInput)对从open方法装饰后的DataInputStream进行反序列化或序列化的,从源码中可以看出Deserialize并未对DataInputStream进行验证,这也就意味着必须先open->Deserialize否则会抛出异常。该类序列化和反序列化调用的实际是Writable的方法,进行了便捷的封装。

结构如下:

 

SerilizationFactory:维护一个SerilizationArrayList。它具有参数为Configuration的构造函数,把parameter io.serializations中逗号隔开的serialization都添加进来。 

结构如下:

说明:

1.该类是SerializationFactory模式,Configuration的衍生类,成员变量为LogArrayList<Serialization<?>>,ArrayList保存SerializationClass集合。

2.构造方法从Conf配置文件中读取io.serializations的属性配置,然后利用add()方法添加到ArrayList<Serialization<?>>中。

3.add()方法从Conf文件利用serializationNam 获得conf.getClassByName(serializationName)Class属性,再利用反射机制实例化该类并添加到ArrayList<Serialization<?>> serializations中。

4.getSerialization方法根据ClassArrayList获取具体的Serialization类。

5.怎么知道对应的Class获取的是哪个序列化框架提供的序列化实例实际上也是通过ArrayList<Serialization<?>>中去获取满足accept()方法所对应的Serialization。

------------------我表示我是分隔符----------------------------------------------

org.apache.hadoop.io中有两个类: 

Stringifier是把类转化为字符串和把字符串转化为类的一个接口。DefaultStringifier实现了这个接口,其中用到的序列化方式就是SerializationFactory中维护的Serialization

下面的程序演示了如何在把Java的序列化方式加入到这个框架中来: 

package test;

import java.io.Serializable;

public class TestSerializer implements Serializable{

	/**
	 * 
	 */
	private static final long serialVersionUID = -5063738225407612355L;
	private int a;
	private String b;
	public TestSerializer(int a, String b) {
		super();
		this.a = a;
		this.b = b;
	}
	public int getA() {
		return a;
	}
	public void setA(int a) {
		this.a = a;
	}
	public String getB() {
		return b;
	}
	public void setB(String b) {
		this.b = b;
	}
}



package test;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.DefaultStringifier;

public class serializer {

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub

		Configuration conf = new Configuration();
		conf.set("io.serializations", "org.apache.hadoop.io.serializer.JavaSerialization,org.apache.hadoop.io.serializer.WritableSerialization");
		TestSerializer ts = new TestSerializer(1,"234");
		DefaultStringifier<TestSerializer> ds = new DefaultStringifier<TestSerializer>(conf, TestSerializer.class);
		String s = null;
		try {
			s = ds.toString(ts);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		System.out.println(s);
		TestSerializer tsxp = null;
		try {
			tsxp = ds.fromString(s);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		System.out.println(tsxp.getA()+":"+tsxp.getB());
	}

}


展开阅读全文
打赏
0
0 收藏
分享
加载中
更多评论
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部