You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ma...@apache.org on 2013/11/10 18:24:05 UTC
[1/3] git commit: Call Kryo setReferences before calling user
specified Kryo registrator.
Updated Branches:
refs/heads/master 3efc01956 -> 58d4f6c8a
Call Kryo setReferences before calling user specified Kryo registrator.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/7c5f70d8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/7c5f70d8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/7c5f70d8
Branch: refs/heads/master
Commit: 7c5f70d8739126d08f48fc6219421e4aea60cacd
Parents: 3efc019
Author: Reynold Xin <rx...@apache.org>
Authored: Sat Nov 9 22:43:36 2013 -0800
Committer: Reynold Xin <rx...@apache.org>
Committed: Sat Nov 9 22:43:36 2013 -0800
----------------------------------------------------------------------
.../apache/spark/serializer/KryoSerializer.scala | 16 ++++++++++------
1 file changed, 10 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/7c5f70d8/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala b/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
index 55b25f1..4f60f0b 100644
--- a/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
+++ b/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
@@ -30,10 +30,14 @@ import org.apache.spark.broadcast.HttpBroadcast
import org.apache.spark.storage.{GetBlock,GotBlock, PutBlock, StorageLevel, TestBlockId}
/**
- * A Spark serializer that uses the [[http://code.google.com/p/kryo/wiki/V1Documentation Kryo 1.x library]].
+ * A Spark serializer that uses the
+ * [[http://code.google.com/p/kryo/wiki/V1Documentation Kryo 1.x library]].
*/
class KryoSerializer extends org.apache.spark.serializer.Serializer with Logging {
- private val bufferSize = System.getProperty("spark.kryoserializer.buffer.mb", "2").toInt * 1024 * 1024
+
+ private val bufferSize = {
+ System.getProperty("spark.kryoserializer.buffer.mb", "2").toInt * 1024 * 1024
+ }
def newKryoOutput() = new KryoOutput(bufferSize)
@@ -42,6 +46,10 @@ class KryoSerializer extends org.apache.spark.serializer.Serializer with Logging
val kryo = instantiator.newKryo()
val classLoader = Thread.currentThread.getContextClassLoader
+ // Allow disabling Kryo reference tracking if user knows their object graphs don't have loops.
+ // Do this before we invoke the user registrator so the user registrator can override this.
+ kryo.setReferences(System.getProperty("spark.kryo.referenceTracking", "true").toBoolean)
+
val blockId = TestBlockId("1")
// Register some commonly used classes
val toRegister: Seq[AnyRef] = Seq(
@@ -78,10 +86,6 @@ class KryoSerializer extends org.apache.spark.serializer.Serializer with Logging
new AllScalaRegistrar().apply(kryo)
kryo.setClassLoader(classLoader)
-
- // Allow disabling Kryo reference tracking if user knows their object graphs don't have loops
- kryo.setReferences(System.getProperty("spark.kryo.referenceTracking", "true").toBoolean)
-
kryo
}
[3/3] git commit: Merge pull request #157 from rxin/kryo
Posted by ma...@apache.org.
Merge pull request #157 from rxin/kryo
3 Kryo related changes.
1. Call Kryo setReferences before calling user specified Kryo registrator. This is done so the user specified registrator can override the default setting.
2. Register more internal classes (MapStatus, BlockManagerId).
3. Slightly refactored the internal class registration to allocate less memory.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/58d4f6c8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/58d4f6c8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/58d4f6c8
Branch: refs/heads/master
Commit: 58d4f6c8a5d9739dc2a3f26f116528457336f0d3
Parents: 3efc019 c845611
Author: Matei Zaharia <ma...@eecs.berkeley.edu>
Authored: Sun Nov 10 09:23:56 2013 -0800
Committer: Matei Zaharia <ma...@eecs.berkeley.edu>
Committed: Sun Nov 10 09:23:56 2013 -0800
----------------------------------------------------------------------
.../spark/serializer/KryoSerializer.scala | 52 +++++++++++---------
1 file changed, 30 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
[2/3] git commit: Moved the Spark internal class registration for
Kryo into an object, and added more classes (e.g. MapStatus,
BlockManagerId) to the registration.
Posted by ma...@apache.org.
Moved the Spark internal class registration for Kryo into an object, and added more classes (e.g. MapStatus, BlockManagerId) to the registration.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/c845611f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/c845611f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/c845611f
Branch: refs/heads/master
Commit: c845611fc387207a51a063d0809770f23fb7b8cc
Parents: 7c5f70d
Author: Reynold Xin <rx...@apache.org>
Authored: Sat Nov 9 23:00:08 2013 -0800
Committer: Reynold Xin <rx...@apache.org>
Committed: Sat Nov 9 23:00:08 2013 -0800
----------------------------------------------------------------------
.../spark/serializer/KryoSerializer.scala | 40 +++++++++++---------
1 file changed, 22 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c845611f/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala b/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
index 4f60f0b..e748c22 100644
--- a/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
+++ b/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
@@ -27,11 +27,11 @@ import com.twitter.chill.{EmptyScalaKryoInstantiator, AllScalaRegistrar}
import org.apache.spark.{SerializableWritable, Logging}
import org.apache.spark.broadcast.HttpBroadcast
-import org.apache.spark.storage.{GetBlock,GotBlock, PutBlock, StorageLevel, TestBlockId}
+import org.apache.spark.scheduler.MapStatus
+import org.apache.spark.storage._
/**
- * A Spark serializer that uses the
- * [[http://code.google.com/p/kryo/wiki/V1Documentation Kryo 1.x library]].
+ * A Spark serializer that uses the [[https://code.google.com/p/kryo/ Kryo serialization library]].
*/
class KryoSerializer extends org.apache.spark.serializer.Serializer with Logging {
@@ -50,21 +50,7 @@ class KryoSerializer extends org.apache.spark.serializer.Serializer with Logging
// Do this before we invoke the user registrator so the user registrator can override this.
kryo.setReferences(System.getProperty("spark.kryo.referenceTracking", "true").toBoolean)
- val blockId = TestBlockId("1")
- // Register some commonly used classes
- val toRegister: Seq[AnyRef] = Seq(
- ByteBuffer.allocate(1),
- StorageLevel.MEMORY_ONLY,
- PutBlock(blockId, ByteBuffer.allocate(1), StorageLevel.MEMORY_ONLY),
- GotBlock(blockId, ByteBuffer.allocate(1)),
- GetBlock(blockId),
- 1 to 10,
- 1 until 10,
- 1L to 10L,
- 1L until 10L
- )
-
- for (obj <- toRegister) kryo.register(obj.getClass)
+ for (cls <- KryoSerializer.toRegister) kryo.register(cls)
// Allow sending SerializableWritable
kryo.register(classOf[SerializableWritable[_]], new KryoJavaSerializer())
@@ -169,3 +155,21 @@ private[spark] class KryoSerializerInstance(ks: KryoSerializer) extends Serializ
trait KryoRegistrator {
def registerClasses(kryo: Kryo)
}
+
+private[serializer] object KryoSerializer {
+ // Commonly used classes.
+ private val toRegister: Seq[Class[_]] = Seq(
+ ByteBuffer.allocate(1).getClass,
+ classOf[StorageLevel],
+ classOf[PutBlock],
+ classOf[GotBlock],
+ classOf[GetBlock],
+ classOf[MapStatus],
+ classOf[BlockManagerId],
+ classOf[Array[Byte]],
+ (1 to 10).getClass,
+ (1 until 10).getClass,
+ (1L to 10L).getClass,
+ (1L until 10L).getClass
+ )
+}