You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by rx...@apache.org on 2014/01/01 02:48:38 UTC
[08/20] git commit: Made SerializableHyperLogLog Externalizable and
added Kryo tests
Made SerializableHyperLogLog Externalizable and added Kryo tests
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/2d511ab3
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/2d511ab3
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/2d511ab3
Branch: refs/heads/master
Commit: 2d511ab320a85eccafbb9e51a2183b07114bbaa1
Parents: 13227aa
Author: Hossein Falaki <fa...@gmail.com>
Authored: Fri Oct 18 15:30:45 2013 -0700
Committer: Hossein Falaki <fa...@gmail.com>
Committed: Fri Oct 18 15:30:45 2013 -0700
----------------------------------------------------------------------
.../org/apache/spark/util/SerializableHyperLogLog.scala | 11 ++++++-----
.../apache/spark/serializer/KryoSerializerSuite.scala | 4 ++++
2 files changed, 10 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/2d511ab3/core/src/main/scala/org/apache/spark/util/SerializableHyperLogLog.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/util/SerializableHyperLogLog.scala b/core/src/main/scala/org/apache/spark/util/SerializableHyperLogLog.scala
index 28a8acc..9cfd414 100644
--- a/core/src/main/scala/org/apache/spark/util/SerializableHyperLogLog.scala
+++ b/core/src/main/scala/org/apache/spark/util/SerializableHyperLogLog.scala
@@ -17,26 +17,27 @@
package org.apache.spark.util
-import java.io.{ObjectOutputStream, ObjectInputStream}
+import java.io.{Externalizable, ObjectOutput, ObjectInput}
import com.clearspring.analytics.stream.cardinality.{ICardinality, HyperLogLog}
/**
- * A wrapper around com.clearspring.analytics.stream.cardinality.HyperLogLog that is serializable.
+ * A wrapper around [[com.clearspring.analytics.stream.cardinality.HyperLogLog]] that is serializable.
*/
private[spark]
-class SerializableHyperLogLog(@transient var value: ICardinality) extends Serializable {
+class SerializableHyperLogLog(var value: ICardinality) extends Externalizable {
+ def this() = this(null) // For deserialization
def merge(other: SerializableHyperLogLog) = new SerializableHyperLogLog(value.merge(other.value))
- private def readObject(in: ObjectInputStream) {
+ def readExternal(in: ObjectInput) {
val byteLength = in.readInt()
val bytes = new Array[Byte](byteLength)
in.readFully(bytes)
value = HyperLogLog.Builder.build(bytes)
}
- private def writeObject(out: ObjectOutputStream) {
+ def writeExternal(out: ObjectOutput) {
val bytes = value.getBytes()
out.writeInt(bytes.length)
out.write(bytes)
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/2d511ab3/core/src/test/scala/org/apache/spark/serializer/KryoSerializerSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/serializer/KryoSerializerSuite.scala b/core/src/test/scala/org/apache/spark/serializer/KryoSerializerSuite.scala
index c016c51..1852971 100644
--- a/core/src/test/scala/org/apache/spark/serializer/KryoSerializerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/serializer/KryoSerializerSuite.scala
@@ -172,6 +172,10 @@ class KryoSerializerSuite extends FunSuite with SharedSparkContext {
assert (sc.parallelize( Array((1, 11), (2, 22), (3, 33)) ).collect().head === (1, 11))
}
+ test("kryo with SerializableHyperLogLog") {
+ assert(sc.parallelize( Array(1, 2, 3, 2, 3, 3, 2, 3, 1) ).countDistinct(0.01) === 3)
+ }
+
test("kryo with reduce") {
val control = 1 :: 2 :: Nil
val result = sc.parallelize(control, 2).map(new ClassWithoutNoArgConstructor(_))