Permalink
Browse files

Added an option (spark.closure.serializer) to specify the serializer for

closures. This enables using Kryo as the closure serializer.
  • Loading branch information...
1 parent 8c95a85 commit 968f75f6afc1383692f4f33d6a1a5f8ce2ac951d @rxin rxin committed Apr 10, 2012
@@ -126,6 +126,10 @@ class WPRSerializerInstance extends SerializerInstance {
throw new UnsupportedOperationException()
}
+ def deserialize[T](bytes: Array[Byte], loader: ClassLoader): T = {
+ throw new UnsupportedOperationException()
+ }
+
def outputStream(s: OutputStream): SerializationStream = {
new WPRSerializationStream(s)
}
@@ -34,6 +34,15 @@ class JavaSerializerInstance extends SerializerInstance {
in.readObject().asInstanceOf[T]
}
+ def deserialize[T](bytes: Array[Byte], loader: ClassLoader): T = {
+ val bis = new ByteArrayInputStream(bytes)
+ val ois = new ObjectInputStream(bis) {
+ override def resolveClass(desc: ObjectStreamClass) =
+ Class.forName(desc.getName, false, loader)
+ }
+ return ois.readObject.asInstanceOf[T]
+ }
+
def outputStream(s: OutputStream): SerializationStream = {
new JavaSerializationStream(s)
}
@@ -9,6 +9,7 @@ import scala.collection.mutable
import com.esotericsoftware.kryo._
import com.esotericsoftware.kryo.{Serializer => KSerializer}
+import com.esotericsoftware.kryo.serialize.ClassSerializer
import de.javakaffee.kryoserializers.KryoReflectionFactorySupport
/**
@@ -100,6 +101,14 @@ class KryoSerializerInstance(ks: KryoSerializer) extends SerializerInstance {
buf.readClassAndObject(bytes).asInstanceOf[T]
}
+ def deserialize[T](bytes: Array[Byte], loader: ClassLoader): T = {
+ val oldClassLoader = ks.kryo.getClassLoader
+ ks.kryo.setClassLoader(loader)
+ val obj = buf.readClassAndObject(bytes).asInstanceOf[T]
+ ks.kryo.setClassLoader(oldClassLoader)
+ obj
+ }
+
def outputStream(s: OutputStream): SerializationStream = {
new KryoSerializationStream(ks.kryo, ks.threadByteBuf.get(), s)
}
@@ -129,6 +138,8 @@ class KryoSerializer extends Serializer with Logging {
}
def createKryo(): Kryo = {
+ // This is used so we can serialize/deserialize objects without a zero-arg
+ // constructor.
val kryo = new KryoReflectionFactorySupport()
// Register some commonly used classes
@@ -150,6 +161,10 @@ class KryoSerializer extends Serializer with Logging {
kryo.register(obj.getClass)
}
+ // Register the following classes for passing closures.
+ kryo.register(classOf[Class[_]], new ClassSerializer(kryo))
+ kryo.setRegistrationOptional(true)
+
// Register some commonly used Scala singleton objects. Because these
// are singletons, we must return the exact same local object when we
// deserialize rather than returning a clone as FieldSerializer would.
@@ -16,6 +16,7 @@ trait Serializer {
trait SerializerInstance {
def serialize[T](t: T): Array[Byte]
def deserialize[T](bytes: Array[Byte]): T
+ def deserialize[T](bytes: Array[Byte], loader: ClassLoader): T
def outputStream(s: OutputStream): SerializationStream
def inputStream(s: InputStream): DeserializationStream
}
@@ -12,27 +12,17 @@ import scala.util.Random
* Various utility methods used by Spark.
*/
object Utils {
- def serialize[T](o: T): Array[Byte] = {
- val bos = new ByteArrayOutputStream()
- val oos = new ObjectOutputStream(bos)
- oos.writeObject(o)
- oos.close
- return bos.toByteArray
- }
- def deserialize[T](bytes: Array[Byte]): T = {
- val bis = new ByteArrayInputStream(bytes)
- val ois = new ObjectInputStream(bis)
- return ois.readObject.asInstanceOf[T]
- }
+ // The serializer in this object is used by Spark to serialize closures.
+ val serializerClass = System.getProperty("spark.closure.serializer", "spark.JavaSerializer")
+ val ser = Class.forName(serializerClass).newInstance().asInstanceOf[Serializer]
+
+ def serialize[T](o: T): Array[Byte] = ser.newInstance().serialize[T](o)
+
+ def deserialize[T](bytes: Array[Byte]): T = ser.newInstance().deserialize[T](bytes)
def deserialize[T](bytes: Array[Byte], loader: ClassLoader): T = {
- val bis = new ByteArrayInputStream(bytes)
- val ois = new ObjectInputStream(bis) {
- override def resolveClass(desc: ObjectStreamClass) =
- Class.forName(desc.getName, false, loader)
- }
- return ois.readObject.asInstanceOf[T]
+ ser.newInstance().deserialize[T](bytes, loader)
}
def isAlpha(c: Char): Boolean = {

0 comments on commit 968f75f

Please sign in to comment.