svn commit: r1662775 - in /hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec: Utilities.java spark/KryoSerializer.java

[xu...@apache.org]

Author: xuefu
Date: Fri Feb 27 18:24:46 2015
New Revision: 1662775

URL: http://svn.apache.org/r1662775
Log:
HIVE-9781: Utilize spark.kryo.classesToRegister [Spark Branch] (Jimmy via Xuefu)

Modified:
    hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
    hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/KryoSerializer.java

Modified: hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
URL: http://svn.apache.org/viewvc/hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java?rev=1662775&r1=1662774&r2=1662775&view=diff
==============================================================================
--- hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java (original)
+++ hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java Fri Feb 27 18:24:46 2015
@@ -83,6 +83,7 @@ import org.apache.commons.codec.binary.B
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang.WordUtils;
 import org.apache.commons.lang3.StringEscapeUtils;
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
@@ -151,6 +152,8 @@ import org.apache.hadoop.hive.ql.plan.Pa
 import org.apache.hadoop.hive.ql.plan.PlanUtils;
 import org.apache.hadoop.hive.ql.plan.PlanUtils.ExpressionTypes;
 import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.SparkEdgeProperty;
+import org.apache.hadoop.hive.ql.plan.SparkWork;
 import org.apache.hadoop.hive.ql.plan.TableDesc;
 import org.apache.hadoop.hive.ql.plan.api.Adjacency;
 import org.apache.hadoop.hive.ql.plan.api.Graph;
@@ -1090,6 +1093,28 @@ public final class Utilities {
     fld.removeField(fieldName);
     kryo.register(type, fld);
   }
+
+  public static ThreadLocal<Kryo> sparkSerializationKryo = new ThreadLocal<Kryo>() {
+    @Override
+    protected synchronized Kryo initialValue() {
+      Kryo kryo = new Kryo();
+      kryo.setClassLoader(Thread.currentThread().getContextClassLoader());
+      kryo.register(java.sql.Date.class, new SqlDateSerializer());
+      kryo.register(java.sql.Timestamp.class, new TimestampSerializer());
+      kryo.register(Path.class, new PathSerializer());
+      kryo.setInstantiatorStrategy(new StdInstantiatorStrategy());
+      removeField(kryo, Operator.class, "colExprMap");
+      removeField(kryo, ColumnInfo.class, "objectInspector");
+      kryo.register(SparkEdgeProperty.class);
+      kryo.register(MapWork.class);
+      kryo.register(ReduceWork.class);
+      kryo.register(SparkWork.class);
+      kryo.register(TableDesc.class);
+      kryo.register(Pair.class);
+      return kryo;
+    };
+  };
+
   private static ThreadLocal<Kryo> cloningQueryPlanKryo = new ThreadLocal<Kryo>() {
     @Override
     protected synchronized Kryo initialValue() {

Modified: hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/KryoSerializer.java
URL: http://svn.apache.org/viewvc/hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/KryoSerializer.java?rev=1662775&r1=1662774&r2=1662775&view=diff
==============================================================================
--- hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/KryoSerializer.java (original)
+++ hive/branches/spark/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/KryoSerializer.java Fri Feb 27 18:24:46 2015
@@ -39,14 +39,14 @@ public class KryoSerializer {
     ByteArrayOutputStream stream = new ByteArrayOutputStream();
     Output output = new Output(stream);
 
-    Utilities.runtimeSerializationKryo.get().writeObject(output, object);
+    Utilities.sparkSerializationKryo.get().writeObject(output, object);
 
     output.close(); // close() also calls flush()
     return stream.toByteArray();
   }
 
   public static <T> T deserialize(byte[] buffer, Class<T> clazz) {
-    return Utilities.runtimeSerializationKryo.get().readObject(
+    return Utilities.sparkSerializationKryo.get().readObject(
         new Input(new ByteArrayInputStream(buffer)), clazz);
   }