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 2015/04/10 21:09:59 UTC
spark git commit: [SQL] [SPARK-6794] Use kryo-based
SparkSqlSerializer for GeneralHashedRelation
Repository: spark
Updated Branches:
refs/heads/master 9f5ed99d6 -> b9baa4cd9
[SQL] [SPARK-6794] Use kryo-based SparkSqlSerializer for GeneralHashedRelation
Benchmarking results: http://pastie.org/private/1dneo1mta5zpsw6gmsoeq
Author: Volodymyr Lyubinets <vl...@gmail.com>
Closes #5433 from vlyubin/joins and squashes the following commits:
d70c829 [Volodymyr Lyubinets] Addressed review feedback
527eac6 [Volodymyr Lyubinets] Use kryo-based SparkSqlSerializer for GeneralHashedRelation
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/b9baa4cd
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/b9baa4cd
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/b9baa4cd
Branch: refs/heads/master
Commit: b9baa4cd9f6e9fc58161f79744b6b7729894d920
Parents: 9f5ed99
Author: Volodymyr Lyubinets <vl...@gmail.com>
Authored: Fri Apr 10 12:09:54 2015 -0700
Committer: Michael Armbrust <mi...@databricks.com>
Committed: Fri Apr 10 12:09:54 2015 -0700
----------------------------------------------------------------------
.../sql/execution/SparkSqlSerializer.scala | 3 +-
.../sql/execution/joins/HashedRelation.scala | 47 ++++++++++++++++++--
2 files changed, 44 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/b9baa4cd/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlSerializer.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlSerializer.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlSerializer.scala
index 967bd76..347e2f4 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlSerializer.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlSerializer.scala
@@ -26,14 +26,13 @@ import scala.reflect.ClassTag
import com.clearspring.analytics.stream.cardinality.HyperLogLog
import com.esotericsoftware.kryo.io.{Input, Output}
import com.esotericsoftware.kryo.{Serializer, Kryo}
-import com.twitter.chill.{AllScalaRegistrar, ResourcePool}
+import com.twitter.chill.ResourcePool
import org.apache.spark.{SparkEnv, SparkConf}
import org.apache.spark.serializer.{SerializerInstance, KryoSerializer}
import org.apache.spark.sql.catalyst.expressions.GenericRow
import org.apache.spark.util.collection.OpenHashSet
import org.apache.spark.util.MutablePair
-import org.apache.spark.util.Utils
import org.apache.spark.sql.catalyst.expressions.codegen.{IntegerHashSet, LongHashSet}
http://git-wip-us.apache.org/repos/asf/spark/blob/b9baa4cd/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala
index 2fa1cf5..ab84c12 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala
@@ -17,9 +17,11 @@
package org.apache.spark.sql.execution.joins
+import java.io.{ObjectInput, ObjectOutput, Externalizable}
import java.util.{HashMap => JavaHashMap}
import org.apache.spark.sql.catalyst.expressions.{Projection, Row}
+import org.apache.spark.sql.execution.SparkSqlSerializer
import org.apache.spark.util.collection.CompactBuffer
@@ -29,16 +31,43 @@ import org.apache.spark.util.collection.CompactBuffer
*/
private[joins] sealed trait HashedRelation {
def get(key: Row): CompactBuffer[Row]
+
+ // This is a helper method to implement Externalizable, and is used by
+ // GeneralHashedRelation and UniqueKeyHashedRelation
+ protected def writeBytes(out: ObjectOutput, serialized: Array[Byte]): Unit = {
+ out.writeInt(serialized.length) // Write the length of serialized bytes first
+ out.write(serialized)
+ }
+
+ // This is a helper method to implement Externalizable, and is used by
+ // GeneralHashedRelation and UniqueKeyHashedRelation
+ protected def readBytes(in: ObjectInput): Array[Byte] = {
+ val serializedSize = in.readInt() // Read the length of serialized bytes first
+ val bytes = new Array[Byte](serializedSize)
+ in.readFully(bytes)
+ bytes
+ }
}
/**
* A general [[HashedRelation]] backed by a hash map that maps the key into a sequence of values.
*/
-private[joins] final class GeneralHashedRelation(hashTable: JavaHashMap[Row, CompactBuffer[Row]])
- extends HashedRelation with Serializable {
+private[joins] final class GeneralHashedRelation(
+ private var hashTable: JavaHashMap[Row, CompactBuffer[Row]])
+ extends HashedRelation with Externalizable {
+
+ def this() = this(null) // Needed for serialization
override def get(key: Row): CompactBuffer[Row] = hashTable.get(key)
+
+ override def writeExternal(out: ObjectOutput): Unit = {
+ writeBytes(out, SparkSqlSerializer.serialize(hashTable))
+ }
+
+ override def readExternal(in: ObjectInput): Unit = {
+ hashTable = SparkSqlSerializer.deserialize(readBytes(in))
+ }
}
@@ -46,8 +75,10 @@ private[joins] final class GeneralHashedRelation(hashTable: JavaHashMap[Row, Com
* A specialized [[HashedRelation]] that maps key into a single value. This implementation
* assumes the key is unique.
*/
-private[joins] final class UniqueKeyHashedRelation(hashTable: JavaHashMap[Row, Row])
- extends HashedRelation with Serializable {
+private[joins] final class UniqueKeyHashedRelation(private var hashTable: JavaHashMap[Row, Row])
+ extends HashedRelation with Externalizable {
+
+ def this() = this(null) // Needed for serialization
override def get(key: Row): CompactBuffer[Row] = {
val v = hashTable.get(key)
@@ -55,6 +86,14 @@ private[joins] final class UniqueKeyHashedRelation(hashTable: JavaHashMap[Row, R
}
def getValue(key: Row): Row = hashTable.get(key)
+
+ override def writeExternal(out: ObjectOutput): Unit = {
+ writeBytes(out, SparkSqlSerializer.serialize(hashTable))
+ }
+
+ override def readExternal(in: ObjectInput): Unit = {
+ hashTable = SparkSqlSerializer.deserialize(readBytes(in))
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org