Spark的序列化

spark的序列化主要使用了hadoop的writable和java的Serializable。

说到底就是使用hadoop的writable替换了java的默认序列化实现方式。

class SerializableWritable[T <: Writable](@transient var t: T) extends Serializable {

  def value: T = t

  override def toString: String = t.toString

  private def writeObject(out: ObjectOutputStream): Unit = Utils.tryOrIOException {

    out.defaultWriteObject()

    new ObjectWritable(t).write(out)

  }

  private def readObject(in: ObjectInputStream): Unit = Utils.tryOrIOException {

    in.defaultReadObject()

    val ow = new ObjectWritable()

    ow.setConf(new Configuration())

    ow.readFields(in)

    t = ow.get().asInstanceOf[T]

  }

}

这个有个让人疑惑的地方是使用@transient 表示该值不会被序列化,我做个一个测试是可以的，为什么呢，因为spark这里定制了java的序列化，使用hadoop的序列化方案，同时t是Writable类型没有实现Serializable接口不能被序列化。

object SerializableWritableTest extends App

{

  println("start")

  val name:Text=new Text("gaoxing");

  val s=new SerializableWritable[Text](name);

  val fout=new FileOutputStream("name.dat");

  val  out=new ObjectOutputStream(fout);

  out.writeObject(s)

  val fin=new FileInputStream("name.dat");

  val in=new ObjectInputStream(fin);

  val n=in.readObject()

  println(n.asInstanceOf[SerializableWritable[Text]].value.toString)

  println("end")

}

巴特西

Spark的序列化

最新文章

热门文章