kylin-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From shaofeng...@apache.org
Subject [3/6] kylin git commit: spark cubing init commit
Date Sun, 25 Dec 2016 14:49:19 GMT
spark cubing init commit


Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/78e6cd5b
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/78e6cd5b
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/78e6cd5b

Branch: refs/heads/sparkcubing-rebase
Commit: 78e6cd5b771d96744e379e66913277e54632552b
Parents: 73fbbb9
Author: shaofengshi <shaofengshi@apache.org>
Authored: Sun Dec 25 15:59:16 2016 +0800
Committer: shaofengshi <shaofengshi@apache.org>
Committed: Sun Dec 25 19:17:32 2016 +0800

----------------------------------------------------------------------
 assembly/pom.xml                                |   5 +-
 .../main/config/assemblies/source-assembly.xml  |   3 -
 .../java/org/apache/kylin/job/DeployUtil.java   |   1 -
 build/conf/kylin-spark-conf.properties          |  27 ++
 .../apache/kylin/common/KylinConfigBase.java    |  38 +-
 .../kylin/common/persistence/ResourceStore.java |  35 ++
 .../org/apache/kylin/common/util/Array.java     |   2 +-
 .../apache/kylin/common/util/SplittedBytes.java |   2 +-
 .../java/org/apache/kylin/cube/CubeSegment.java |   2 +-
 .../kylin/cube/common/RowKeySplitter.java       |  12 +-
 .../org/apache/kylin/cube/cuboid/Cuboid.java    |  27 +-
 .../kylin/cube/cuboid/CuboidScheduler.java      |  15 +-
 .../kylin/cube/kv/AbstractRowKeyEncoder.java    |   6 +-
 .../org/apache/kylin/cube/kv/CubeDimEncMap.java |   7 +-
 .../apache/kylin/cube/kv/RowKeyColumnIO.java    |   2 +-
 .../org/apache/kylin/cube/kv/RowKeyEncoder.java |  19 +-
 .../kylin/cube/model/AggregationGroup.java      |   4 +-
 .../org/apache/kylin/cube/model/CubeDesc.java   |   2 +-
 .../cube/model/CubeJoinedFlatTableDesc.java     |   2 +-
 .../cube/model/CubeJoinedFlatTableEnrich.java   |   6 +-
 .../apache/kylin/cube/model/DictionaryDesc.java |   2 +-
 .../apache/kylin/cube/model/DimensionDesc.java  |   2 +-
 .../kylin/cube/model/HBaseColumnDesc.java       |  11 +-
 .../kylin/cube/model/HBaseColumnFamilyDesc.java |   9 +-
 .../kylin/cube/model/HBaseMappingDesc.java      |  15 +-
 .../apache/kylin/cube/model/HierarchyDesc.java  |   2 +-
 .../apache/kylin/cube/model/RowKeyColDesc.java  |   2 +-
 .../org/apache/kylin/cube/model/RowKeyDesc.java |  17 +-
 .../org/apache/kylin/cube/model/SelectRule.java |   2 +-
 .../dict/NumberDictionaryForestBuilder.java     |   2 +-
 .../apache/kylin/dict/StringBytesConverter.java |   2 +-
 .../kylin/job/execution/ExecutableManager.java  |  21 +-
 .../kylin/dimension/AbstractDateDimEnc.java     |  12 +-
 .../apache/kylin/dimension/BooleanDimEnc.java   |   4 +-
 .../kylin/dimension/DictionaryDimEnc.java       |  21 +-
 .../apache/kylin/dimension/FixedLenDimEnc.java  |   4 +-
 .../kylin/dimension/FixedLenHexDimEnc.java      |   4 +-
 .../org/apache/kylin/dimension/IntDimEnc.java   |   4 +-
 .../apache/kylin/dimension/IntegerDimEnc.java   |   4 +-
 .../kylin/dimension/OneMoreByteVLongDimEnc.java |  14 +-
 .../kylin/measure/BufferedMeasureCodec.java     |  10 +-
 .../org/apache/kylin/measure/MeasureCodec.java  |   8 +-
 .../apache/kylin/measure/MeasureIngester.java   |  12 +-
 .../org/apache/kylin/measure/MeasureType.java   |  12 +-
 .../kylin/measure/basic/DoubleIngester.java     |   5 +
 .../measure/basic/DoubleSumAggregator.java      |   3 +-
 .../kylin/measure/basic/LongIngester.java       |   5 +
 .../kylin/measure/basic/LongSumAggregator.java  |   3 +-
 .../kylin/measure/bitmap/BitmapCounter.java     |   8 +-
 .../kylin/measure/bitmap/BitmapMeasureType.java |   5 +
 .../kylin/measure/hllc/HLLCMeasureType.java     |   5 +
 .../kylin/measure/hllc/HLLCSerializer.java      |  11 +-
 .../apache/kylin/measure/raw/RawSerializer.java |   4 +-
 .../apache/kylin/measure/topn/TopNCounter.java  |   2 +-
 .../kylin/measure/topn/TopNMeasureType.java     |   4 +-
 .../metadata/datatype/BigDecimalSerializer.java |   2 +-
 .../metadata/datatype/BooleanSerializer.java    |   9 +-
 .../metadata/datatype/DataTypeSerializer.java   |  17 +-
 .../metadata/datatype/DateTimeSerializer.java   |   9 +-
 .../metadata/datatype/DoubleSerializer.java     |   5 +-
 .../kylin/metadata/datatype/Int4Serializer.java |   9 +-
 .../metadata/datatype/Long8Serializer.java      |   9 +-
 .../kylin/metadata/datatype/LongSerializer.java |   9 +-
 .../kylin/metadata/model/FunctionDesc.java      |  22 +-
 .../apache/kylin/metadata/model/JoinDesc.java   |   7 +-
 .../kylin/metadata/model/JoinTableDesc.java     |   4 +-
 .../apache/kylin/metadata/model/JoinsTree.java  |   9 +-
 .../kylin/metadata/model/MeasureDesc.java       |   9 +-
 .../metadata/model/ModelDimensionDesc.java      |  10 +-
 .../kylin/metadata/model/ParameterDesc.java     |  13 +-
 .../kylin/metadata/model/PartitionDesc.java     |   6 +-
 .../apache/kylin/metadata/model/TableRef.java   |  17 +-
 .../kylin/engine/mr/BatchCubingJobBuilder2.java |   9 +-
 .../kylin/engine/mr/JobBuilderSupport.java      |  12 +-
 .../engine/mr/common/BaseCuboidBuilder.java     | 175 +++++++++
 .../kylin/engine/mr/common/NDCuboidBuilder.java |  96 +++++
 .../engine/mr/steps/BaseCuboidMapperBase.java   | 131 +------
 .../kylin/engine/mr/steps/NDCuboidMapper.java   |  58 +--
 engine-spark/pom.xml                            |   5 +
 .../engine/spark/SparkBatchCubingEngine2.java   |  33 ++
 .../spark/SparkBatchCubingJobBuilder2.java      |  85 +++++
 .../apache/kylin/engine/spark/SparkCubing.java  |  79 ++++-
 .../kylin/engine/spark/SparkCubingV3.java       | 354 +++++++++++++++++++
 .../kylin/engine/spark/SparkExecutable.java     |  26 +-
 .../sandbox/kylin-spark-conf.properties         |  27 ++
 .../test_case_data/sandbox/kylin.properties     |  11 +-
 pom.xml                                         |   2 +-
 .../kylin/rest/controller/CubeController.java   |   8 +-
 server/pom.xml                                  |   7 +
 .../apache/kylin/source/hive/HiveMRInput.java   |   5 +-
 tool/pom.xml                                    |   4 +
 91 files changed, 1299 insertions(+), 462 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/assembly/pom.xml
----------------------------------------------------------------------
diff --git a/assembly/pom.xml b/assembly/pom.xml
index 73f9e12..fc49a62 100644
--- a/assembly/pom.xml
+++ b/assembly/pom.xml
@@ -47,6 +47,10 @@
             <groupId>org.apache.kylin</groupId>
             <artifactId>kylin-engine-mr</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.kylin</groupId>
+            <artifactId>kylin-engine-spark</artifactId>
+        </dependency>
 
         <!-- Env & Test -->
         <dependency>
@@ -173,7 +177,6 @@
                             <shadedClassifierName>job</shadedClassifierName>
                             <artifactSet>
                                 <excludes>
-                                    <exclude>io.netty:*</exclude>
                                     <exclude>org.apache.zookeeper:*</exclude>
                                     <exclude>net.sf.ehcache:*</exclude>
                                     <exclude>org.apache.httpcomponents:*</exclude>

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/assembly/src/main/config/assemblies/source-assembly.xml
----------------------------------------------------------------------
diff --git a/assembly/src/main/config/assemblies/source-assembly.xml b/assembly/src/main/config/assemblies/source-assembly.xml
index fad45aa..92584b5 100644
--- a/assembly/src/main/config/assemblies/source-assembly.xml
+++ b/assembly/src/main/config/assemblies/source-assembly.xml
@@ -97,9 +97,6 @@ limitations under the License.
                 <exclude>%regex[(?!((?!${project.build.directory}/)[^/]+/)*src/)(.*/)?docs(/.*)?]
                 </exclude>
 
-                <!-- exclude unmaintained -->
-                <exclude>%regex[(?!((?!${project.build.directory}/)))?engine-spark(/.*)?]
-                </exclude>
             </excludes>
         </fileSet>
         <!-- LICENSE, NOTICE, DEPENDENCIES, git.properties, etc. calculated at build time -->

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/assembly/src/test/java/org/apache/kylin/job/DeployUtil.java
----------------------------------------------------------------------
diff --git a/assembly/src/test/java/org/apache/kylin/job/DeployUtil.java b/assembly/src/test/java/org/apache/kylin/job/DeployUtil.java
index 089c3ed..d1abfeb 100644
--- a/assembly/src/test/java/org/apache/kylin/job/DeployUtil.java
+++ b/assembly/src/test/java/org/apache/kylin/job/DeployUtil.java
@@ -82,7 +82,6 @@ public class DeployUtil {
 
         config().overrideMRJobJarPath(jobJar.getAbsolutePath());
         config().overrideCoprocessorLocalJar(coprocessorJar.getAbsolutePath());
-        config().overrideSparkJobJarPath(getSparkJobJarFile().getAbsolutePath());
     }
 
     private static String getPomVersion() {

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/build/conf/kylin-spark-conf.properties
----------------------------------------------------------------------
diff --git a/build/conf/kylin-spark-conf.properties b/build/conf/kylin-spark-conf.properties
new file mode 100644
index 0000000..7f53f50
--- /dev/null
+++ b/build/conf/kylin-spark-conf.properties
@@ -0,0 +1,27 @@
+spark.yarn.submit.file.replication=1
+spark.yarn.executor.memoryOverhead=200
+spark.yarn.driver.memoryOverhead=384
+#spark.master=local[4]
+spark.master=yarn
+spark.submit.deployMode=cluster
+spark.eventLog.enabled=true
+spark.yarn.scheduler.heartbeat.interval-ms=5000
+spark.yarn.preserve.staging.files=true
+spark.yarn.queue=default
+spark.yarn.containerLauncherMaxThreads=25
+spark.yarn.max.executor.failures=3
+spark.eventLog.dir=hdfs\:///kylin/spark-history
+spark.history.kerberos.enabled=true
+spark.history.provider=org.apache.spark.deploy.history.FsHistoryProvider
+spark.history.ui.port=18080
+spark.history.fs.logDirectory=hdfs\:///kylin/spark-history
+spark.executor.memory=1G
+spark.storage.memoryFraction=0.3
+spark.executor.cores=1
+spark.executor.instances=1
+spark.history.kerberos.keytab=none
+spark.history.kerberos.principal=none
+#spark.yarn.jar=hdfs://sandbox.hortonworks.com:8020/apps/spark/spark-assembly-1.6.3-hadoop2.6.0.jar
+spark.driver.extraJavaOptions=-Dhdp.version=current
+spark.yarn.am.extraJavaOptions=-Dhdp.version=current
+spark.executor.extraJavaOptions=-Dhdp.version=current

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
index da93388..c32550f 100644
--- a/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
+++ b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
@@ -636,6 +636,7 @@ abstract public class KylinConfigBase implements Serializable {
         // ref constants in IEngineAware
         r.put(0, "org.apache.kylin.engine.mr.MRBatchCubingEngine");
         r.put(2, "org.apache.kylin.engine.mr.MRBatchCubingEngine2");
+        r.put(4, "org.apache.kylin.engine.spark.SparkBatchCubingEngine2");
         return r;
     }
 
@@ -712,29 +713,32 @@ abstract public class KylinConfigBase implements Serializable {
     // ENGINE.SPARK
     // ============================================================================
 
-    public String getKylinSparkJobJarPath() {
-        final String jobJar = getOptional("kylin.engine.spark.job-jar");
-        if (StringUtils.isNotEmpty(jobJar)) {
-            return jobJar;
-        }
-        String kylinHome = getKylinHome();
-        if (StringUtils.isEmpty(kylinHome)) {
-            return "";
-        }
-        return getFileName(kylinHome + File.separator + "lib", SPARK_JOB_JAR_NAME_PATTERN);
+    public String getSparkHome() {
+        return getRequired("kylin.engine.spark.spark-home");
     }
 
-    public void overrideSparkJobJarPath(String path) {
-        logger.info("override " + "kylin.engine.spark.job-jar" + " to " + path);
-        System.setProperty("kylin.engine.spark.job-jar", path);
+    public String getSparkHadoopConfDir() {
+        return getRequired("kylin.engine.spark.env.hadoop-conf-dir");
     }
 
-    public String getSparkHome() {
-        return getRequired("kylin.engine.spark.spark-home");
+    public String getSparkConfFile() {
+        String conf = getOptional("kylin.engine.spark.properties-file", "conf/kylin-spark-conf.properties");
+        File f = new File(conf);
+        if (f.exists()) {
+            return f.getAbsolutePath();
+        } else {
+            String home = getKylinHome();
+            f = new File(home, conf);
+            if (f.exists()) {
+                return f.getAbsolutePath();
+            }
+        }
+
+        throw new IllegalArgumentException("Spark conf file '" + conf + "' does not exist.");
     }
 
-    public String getSparkMaster() {
-        return getRequired("kylin.engine.spark.spark-master");
+    public String getSparkAdditionalJars() {
+        return getOptional("kylin.engine.spark.additional-jars", "");
     }
 
     // ============================================================================

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-common/src/main/java/org/apache/kylin/common/persistence/ResourceStore.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/persistence/ResourceStore.java b/core-common/src/main/java/org/apache/kylin/common/persistence/ResourceStore.java
index 0580576..9549569 100644
--- a/core-common/src/main/java/org/apache/kylin/common/persistence/ResourceStore.java
+++ b/core-common/src/main/java/org/apache/kylin/common/persistence/ResourceStore.java
@@ -21,17 +21,21 @@ package org.apache.kylin.common.persistence;
 import java.io.ByteArrayOutputStream;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
+import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.NavigableSet;
 import java.util.concurrent.ConcurrentHashMap;
 
+import org.apache.commons.io.FileUtils;
 import org.apache.commons.io.IOUtils;
 import org.apache.kylin.common.KylinConfig;
 import org.apache.kylin.common.util.ClassUtil;
+import org.apache.kylin.common.util.OptionsHelper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -311,4 +315,35 @@ abstract public class ResourceStore {
         return collector;
     }
 
+    public static String dumpResources(KylinConfig kylinConfig, Collection<String> dumpList) throws IOException {
+        File tmp = File.createTempFile("kylin_job_meta", "");
+        FileUtils.forceDelete(tmp); // we need a directory, so delete the file first
+
+        File metaDir = new File(tmp, "meta");
+        metaDir.mkdirs();
+
+        // write kylin.properties
+        File kylinPropsFile = new File(metaDir, "kylin.properties");
+        kylinConfig.writeProperties(kylinPropsFile);
+
+        ResourceStore from = ResourceStore.getStore(kylinConfig);
+        KylinConfig localConfig = KylinConfig.createInstanceFromUri(metaDir.getAbsolutePath());
+        ResourceStore to = ResourceStore.getStore(localConfig);
+        for (String path : dumpList) {
+            RawResource res = from.getResource(path);
+            if (res == null)
+                throw new IllegalStateException("No resource found at -- " + path);
+            to.putResource(path, res.inputStream, res.timestamp);
+            res.inputStream.close();
+        }
+
+        String metaDirURI = OptionsHelper.convertToFileURL(metaDir.getAbsolutePath());
+        if (metaDirURI.startsWith("/")) // note Path on windows is like "d:/../..."
+            metaDirURI = "file://" + metaDirURI;
+        else
+            metaDirURI = "file:///" + metaDirURI;
+        logger.info("meta dir is: " + metaDirURI);
+
+        return metaDirURI;
+    }
 }

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-common/src/main/java/org/apache/kylin/common/util/Array.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/util/Array.java b/core-common/src/main/java/org/apache/kylin/common/util/Array.java
index 7447b46..b25b764 100644
--- a/core-common/src/main/java/org/apache/kylin/common/util/Array.java
+++ b/core-common/src/main/java/org/apache/kylin/common/util/Array.java
@@ -23,7 +23,7 @@ import java.util.Arrays;
 /*
  * An array with correct equals(), hashCode(), compareTo() and toString()
  */
-public class Array<T> implements Comparable<Array<T>> {
+public class Array<T> implements Comparable<Array<T>>, java.io.Serializable {
     public T[] data;
 
     public Array(T[] data) {

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-common/src/main/java/org/apache/kylin/common/util/SplittedBytes.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/util/SplittedBytes.java b/core-common/src/main/java/org/apache/kylin/common/util/SplittedBytes.java
index 8751b78..ae380cb 100644
--- a/core-common/src/main/java/org/apache/kylin/common/util/SplittedBytes.java
+++ b/core-common/src/main/java/org/apache/kylin/common/util/SplittedBytes.java
@@ -22,7 +22,7 @@ package org.apache.kylin.common.util;
  * @author George Song (ysong1)
  * 
  */
-public class SplittedBytes {
+public class SplittedBytes implements java.io.Serializable {
     public SplittedBytes(int length) {
         this.value = new byte[length];
         this.length = 0;

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/CubeSegment.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/CubeSegment.java b/core-cube/src/main/java/org/apache/kylin/cube/CubeSegment.java
index 36a6044..45310f0 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/CubeSegment.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/CubeSegment.java
@@ -51,7 +51,7 @@ import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 
 @JsonAutoDetect(fieldVisibility = Visibility.NONE, getterVisibility = Visibility.NONE, isGetterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE)
-public class CubeSegment implements Comparable<CubeSegment>, IBuildable, ISegment {
+public class CubeSegment implements Comparable<CubeSegment>, IBuildable, ISegment, java.io.Serializable {
 
     @JsonBackReference
     private CubeInstance cubeInstance;

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/common/RowKeySplitter.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/common/RowKeySplitter.java b/core-cube/src/main/java/org/apache/kylin/cube/common/RowKeySplitter.java
index 67f1751..3602fff 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/common/RowKeySplitter.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/common/RowKeySplitter.java
@@ -26,9 +26,11 @@ import org.apache.kylin.cube.kv.CubeDimEncMap;
 import org.apache.kylin.cube.kv.RowConstants;
 import org.apache.kylin.cube.kv.RowKeyColumnIO;
 import org.apache.kylin.cube.model.CubeDesc;
+import org.apache.kylin.cube.model.RowKeyColDesc;
+import org.apache.kylin.dimension.IDimensionEncodingMap;
 import org.apache.kylin.metadata.model.TblColRef;
 
-public class RowKeySplitter {
+public class RowKeySplitter implements java.io.Serializable {
 
     private CubeDesc cubeDesc;
     private RowKeyColumnIO colIO;
@@ -64,7 +66,13 @@ public class RowKeySplitter {
     public RowKeySplitter(CubeSegment cubeSeg, int splitLen, int bytesLen) {
         this.enableSharding = cubeSeg.isEnableSharding();
         this.cubeDesc = cubeSeg.getCubeDesc();
-        this.colIO = new RowKeyColumnIO(new CubeDimEncMap(cubeSeg));
+        IDimensionEncodingMap dimEncoding = new CubeDimEncMap(cubeSeg);
+
+        for (RowKeyColDesc rowKeyColDesc : cubeDesc.getRowkey().getRowKeyColumns()) {
+            dimEncoding.get(rowKeyColDesc.getColRef());
+        }
+
+        this.colIO = new RowKeyColumnIO(dimEncoding);
 
         this.splitBuffers = new SplittedBytes[splitLen];
         this.splitOffsets = new int[splitLen];

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/cuboid/Cuboid.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/cuboid/Cuboid.java b/core-cube/src/main/java/org/apache/kylin/cube/cuboid/Cuboid.java
index 7503fbf..dd22d6a 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/cuboid/Cuboid.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/cuboid/Cuboid.java
@@ -18,15 +18,10 @@
 
 package org.apache.kylin.cube.cuboid;
 
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-
+import com.google.common.base.Function;
+import com.google.common.collect.Collections2;
+import com.google.common.collect.ComparisonChain;
+import com.google.common.collect.Lists;
 import org.apache.commons.lang.StringUtils;
 import org.apache.kylin.common.util.Bytes;
 import org.apache.kylin.cube.gridtable.CuboidToGridTableMapping;
@@ -37,12 +32,16 @@ import org.apache.kylin.cube.model.RowKeyColDesc;
 import org.apache.kylin.metadata.model.FunctionDesc;
 import org.apache.kylin.metadata.model.TblColRef;
 
-import com.google.common.base.Function;
-import com.google.common.collect.Collections2;
-import com.google.common.collect.ComparisonChain;
-import com.google.common.collect.Lists;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 
-public class Cuboid implements Comparable<Cuboid> {
+public class Cuboid implements Comparable<Cuboid>, java.io.Serializable {
 
     private final static Map<String, Map<Long, Cuboid>> CUBOID_CACHE = new ConcurrentHashMap<String, Map<Long, Cuboid>>();
 

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/cuboid/CuboidScheduler.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/cuboid/CuboidScheduler.java b/core-cube/src/main/java/org/apache/kylin/cube/cuboid/CuboidScheduler.java
index 733aded..ffb0a5e 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/cuboid/CuboidScheduler.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/cuboid/CuboidScheduler.java
@@ -21,6 +21,12 @@ package org.apache.kylin.cube.cuboid;
 /**
  */
 
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+import org.apache.kylin.cube.model.AggregationGroup;
+import org.apache.kylin.cube.model.CubeDesc;
+
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
@@ -28,14 +34,7 @@ import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
-import org.apache.kylin.cube.model.AggregationGroup;
-import org.apache.kylin.cube.model.CubeDesc;
-
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Sets;
-
-public class CuboidScheduler {
+public class CuboidScheduler implements java.io.Serializable {
 
     private final CubeDesc cubeDesc;
     private final long max;

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/kv/AbstractRowKeyEncoder.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/kv/AbstractRowKeyEncoder.java b/core-cube/src/main/java/org/apache/kylin/cube/kv/AbstractRowKeyEncoder.java
index bfe6eb4..2becde4 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/kv/AbstractRowKeyEncoder.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/kv/AbstractRowKeyEncoder.java
@@ -18,8 +18,6 @@
 
 package org.apache.kylin.cube.kv;
 
-import java.util.Map;
-
 import org.apache.kylin.common.util.ByteArray;
 import org.apache.kylin.common.util.ImmutableBitSet;
 import org.apache.kylin.cube.CubeSegment;
@@ -30,12 +28,14 @@ import org.apache.kylin.metadata.model.TblColRef;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Map;
+
 /**
  * 
  * @author xjiang
  * 
  */
-public abstract class AbstractRowKeyEncoder {
+public abstract class AbstractRowKeyEncoder implements java.io.Serializable {
 
     protected static final Logger logger = LoggerFactory.getLogger(AbstractRowKeyEncoder.class);
     public static final byte DEFAULT_BLANK_BYTE = DimensionEncoding.NULL;

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/kv/CubeDimEncMap.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/kv/CubeDimEncMap.java b/core-cube/src/main/java/org/apache/kylin/cube/kv/CubeDimEncMap.java
index a4d2d6f..bd9554a 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/kv/CubeDimEncMap.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/kv/CubeDimEncMap.java
@@ -18,8 +18,7 @@
 
 package org.apache.kylin.cube.kv;
 
-import java.util.Map;
-
+import com.google.common.collect.Maps;
 import org.apache.kylin.common.util.Dictionary;
 import org.apache.kylin.cube.CubeSegment;
 import org.apache.kylin.cube.model.CubeDesc;
@@ -33,9 +32,9 @@ import org.apache.kylin.metadata.model.TblColRef;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.google.common.collect.Maps;
+import java.util.Map;
 
-public class CubeDimEncMap implements IDimensionEncodingMap {
+public class CubeDimEncMap implements IDimensionEncodingMap, java.io.Serializable {
 
     private static final Logger logger = LoggerFactory.getLogger(CubeDimEncMap.class);
 

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyColumnIO.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyColumnIO.java b/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyColumnIO.java
index fbb93db..65911a0 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyColumnIO.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyColumnIO.java
@@ -29,7 +29,7 @@ import org.apache.kylin.metadata.model.TblColRef;
  *
  * @author yangli9
  */
-public class RowKeyColumnIO {
+public class RowKeyColumnIO implements java.io.Serializable {
 
     //private static final Logger logger = LoggerFactory.getLogger(RowKeyColumnIO.class);
 

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyEncoder.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyEncoder.java b/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyEncoder.java
index bf20de1..a669fb1 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyEncoder.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyEncoder.java
@@ -18,11 +18,7 @@
 
 package org.apache.kylin.cube.kv;
 
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
+import com.google.common.base.Preconditions;
 import org.apache.kylin.common.util.ByteArray;
 import org.apache.kylin.common.util.BytesUtil;
 import org.apache.kylin.common.util.ImmutableBitSet;
@@ -32,9 +28,12 @@ import org.apache.kylin.cube.cuboid.Cuboid;
 import org.apache.kylin.gridtable.GTRecord;
 import org.apache.kylin.metadata.model.TblColRef;
 
-import com.google.common.base.Preconditions;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
-public class RowKeyEncoder extends AbstractRowKeyEncoder {
+public class RowKeyEncoder extends AbstractRowKeyEncoder implements java.io.Serializable {
 
     private int bodyLength = 0;
     private RowKeyColumnIO colIO;
@@ -42,11 +41,13 @@ public class RowKeyEncoder extends AbstractRowKeyEncoder {
     protected boolean enableSharding;
     private int uhcOffset = -1;//it's a offset to the beginning of body
     private int uhcLength = -1;
+    private int headerLength;
 
     public RowKeyEncoder(CubeSegment cubeSeg, Cuboid cuboid) {
         super(cubeSeg, cuboid);
         enableSharding = cubeSeg.isEnableSharding();
-        Set<TblColRef> shardByColumns = cubeSeg.getShardByColumns();
+        headerLength = cubeSeg.getRowKeyPreambleSize();
+        Set<TblColRef> shardByColumns = cubeSeg.getCubeDesc().getShardByColumns();
         if (shardByColumns.size() > 1) {
             throw new IllegalStateException("Does not support multiple UHC now");
         }
@@ -61,7 +62,7 @@ public class RowKeyEncoder extends AbstractRowKeyEncoder {
     }
 
     public int getHeaderLength() {
-        return cubeSeg.getRowKeyPreambleSize();
+        return headerLength;
     }
 
     public int getBytesLength() {

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/model/AggregationGroup.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/AggregationGroup.java b/core-cube/src/main/java/org/apache/kylin/cube/model/AggregationGroup.java
index 9bd082f..5f3c92c 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/AggregationGroup.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/AggregationGroup.java
@@ -35,8 +35,8 @@ import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
 
 @JsonAutoDetect(fieldVisibility = Visibility.NONE, getterVisibility = Visibility.NONE, isGetterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE)
-public class AggregationGroup {
-    public static class HierarchyMask {
+public class AggregationGroup implements java.io.Serializable {
+    public static class HierarchyMask implements java.io.Serializable {
         public long fullMask; // 00000111
         public long[] allMasks; // 00000100,00000110,00000111
         public long[] dims; // 00000100,00000010,00000001

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
index 3b8d034..c9ad8bc 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
@@ -100,7 +100,7 @@ public class CubeDesc extends RootPersistentEntity implements IEngineAware {
         LOOKUP, PK_FK, EXTENDED_COLUMN
     }
 
-    public static class DeriveInfo {
+    public static class DeriveInfo implements java.io.Serializable {
         public DeriveType type;
         public JoinDesc join;
         public TblColRef[] columns;

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeJoinedFlatTableDesc.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/CubeJoinedFlatTableDesc.java b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeJoinedFlatTableDesc.java
index 94e1a7c..2a68116 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/CubeJoinedFlatTableDesc.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeJoinedFlatTableDesc.java
@@ -36,7 +36,7 @@ import com.google.common.collect.Maps;
 
 /**
  */
-public class CubeJoinedFlatTableDesc implements IJoinedFlatTableDesc {
+public class CubeJoinedFlatTableDesc implements IJoinedFlatTableDesc, java.io.Serializable {
 
     protected final String tableName;
     protected final CubeDesc cubeDesc;

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeJoinedFlatTableEnrich.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/CubeJoinedFlatTableEnrich.java b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeJoinedFlatTableEnrich.java
index a1312b5..e829aeb 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/CubeJoinedFlatTableEnrich.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeJoinedFlatTableEnrich.java
@@ -18,8 +18,6 @@
 
 package org.apache.kylin.cube.model;
 
-import java.util.List;
-
 import org.apache.kylin.cube.cuboid.Cuboid;
 import org.apache.kylin.metadata.model.DataModelDesc;
 import org.apache.kylin.metadata.model.FunctionDesc;
@@ -28,10 +26,12 @@ import org.apache.kylin.metadata.model.ISegment;
 import org.apache.kylin.metadata.model.MeasureDesc;
 import org.apache.kylin.metadata.model.TblColRef;
 
+import java.util.List;
+
 /**
  * An enrich of IJoinedFlatTableDesc for cubes
  */
-public class CubeJoinedFlatTableEnrich implements IJoinedFlatTableDesc {
+public class CubeJoinedFlatTableEnrich implements IJoinedFlatTableDesc, java.io.Serializable {
 
     private CubeDesc cubeDesc;
     private IJoinedFlatTableDesc flatDesc;

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/model/DictionaryDesc.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/DictionaryDesc.java b/core-cube/src/main/java/org/apache/kylin/cube/model/DictionaryDesc.java
index f471f9f..ca2183a 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/DictionaryDesc.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/DictionaryDesc.java
@@ -27,7 +27,7 @@ import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
 @JsonAutoDetect(fieldVisibility = Visibility.NONE, getterVisibility = Visibility.NONE, isGetterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE)
-public class DictionaryDesc {
+public class DictionaryDesc implements java.io.Serializable {
 
     @JsonProperty("column")
     private String column;

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/model/DimensionDesc.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/DimensionDesc.java b/core-cube/src/main/java/org/apache/kylin/cube/model/DimensionDesc.java
index cd75228..93bf4bd 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/DimensionDesc.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/DimensionDesc.java
@@ -35,7 +35,7 @@ import com.google.common.base.Objects;
 /**
  */
 @JsonAutoDetect(fieldVisibility = Visibility.NONE, getterVisibility = Visibility.NONE, isGetterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE)
-public class DimensionDesc {
+public class DimensionDesc implements java.io.Serializable {
 
     @JsonProperty("name")
     private String name;

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/model/HBaseColumnDesc.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/HBaseColumnDesc.java b/core-cube/src/main/java/org/apache/kylin/cube/model/HBaseColumnDesc.java
index fb491f8..7007342 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/HBaseColumnDesc.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/HBaseColumnDesc.java
@@ -18,19 +18,18 @@
 
 package org.apache.kylin.cube.model;
 
-import java.util.Arrays;
-
-import org.apache.kylin.metadata.model.FunctionDesc;
-import org.apache.kylin.metadata.model.MeasureDesc;
-
 import com.fasterxml.jackson.annotation.JsonAutoDetect;
 import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.kylin.metadata.model.FunctionDesc;
+import org.apache.kylin.metadata.model.MeasureDesc;
+
+import java.util.Arrays;
 
 /**
  */
 @JsonAutoDetect(fieldVisibility = Visibility.NONE, getterVisibility = Visibility.NONE, isGetterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE)
-public class HBaseColumnDesc {
+public class HBaseColumnDesc implements java.io.Serializable {
 
     @JsonProperty("qualifier")
     private String qualifier;

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/model/HBaseColumnFamilyDesc.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/HBaseColumnFamilyDesc.java b/core-cube/src/main/java/org/apache/kylin/cube/model/HBaseColumnFamilyDesc.java
index c5b2e19..85c2c17 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/HBaseColumnFamilyDesc.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/HBaseColumnFamilyDesc.java
@@ -18,18 +18,17 @@
 
 package org.apache.kylin.cube.model;
 
-import java.util.Arrays;
-
-import org.apache.kylin.metadata.model.MeasureDesc;
-
 import com.fasterxml.jackson.annotation.JsonAutoDetect;
 import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.kylin.metadata.model.MeasureDesc;
+
+import java.util.Arrays;
 
 /**
  */
 @JsonAutoDetect(fieldVisibility = Visibility.NONE, getterVisibility = Visibility.NONE, isGetterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE)
-public class HBaseColumnFamilyDesc {
+public class HBaseColumnFamilyDesc implements java.io.Serializable {
 
     @JsonProperty("name")
     private String name;

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/model/HBaseMappingDesc.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/HBaseMappingDesc.java b/core-cube/src/main/java/org/apache/kylin/cube/model/HBaseMappingDesc.java
index 2ef1e17..9ad8407 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/HBaseMappingDesc.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/HBaseMappingDesc.java
@@ -18,22 +18,21 @@
 
 package org.apache.kylin.cube.model;
 
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.LinkedList;
-
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
+import com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.kylin.common.util.StringUtil;
 import org.apache.kylin.metadata.model.FunctionDesc;
 import org.apache.kylin.metadata.model.MeasureDesc;
 
-import com.fasterxml.jackson.annotation.JsonAutoDetect;
-import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
-import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedList;
 
 /**
  */
 @JsonAutoDetect(fieldVisibility = Visibility.NONE, getterVisibility = Visibility.NONE, isGetterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE)
-public class HBaseMappingDesc {
+public class HBaseMappingDesc implements java.io.Serializable {
 
     @JsonProperty("column_family")
     private HBaseColumnFamilyDesc[] columnFamily;

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/model/HierarchyDesc.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/HierarchyDesc.java b/core-cube/src/main/java/org/apache/kylin/cube/model/HierarchyDesc.java
index f88d4d2..b72f220 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/HierarchyDesc.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/HierarchyDesc.java
@@ -27,7 +27,7 @@ import com.fasterxml.jackson.annotation.JsonProperty;
 /**
  */
 @JsonAutoDetect(fieldVisibility = Visibility.NONE, getterVisibility = Visibility.NONE, isGetterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE)
-public class HierarchyDesc {
+public class HierarchyDesc implements java.io.Serializable {
 
     @JsonProperty("level")
     private String level;

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/model/RowKeyColDesc.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/RowKeyColDesc.java b/core-cube/src/main/java/org/apache/kylin/cube/model/RowKeyColDesc.java
index ef34a9b..71fd4b9 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/RowKeyColDesc.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/RowKeyColDesc.java
@@ -40,7 +40,7 @@ import com.google.common.base.Preconditions;
  * 
  */
 @JsonAutoDetect(fieldVisibility = Visibility.NONE, getterVisibility = Visibility.NONE, isGetterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE)
-public class RowKeyColDesc {
+public class RowKeyColDesc implements java.io.Serializable {
 
     @JsonProperty("column")
     private String column;

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/model/RowKeyDesc.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/RowKeyDesc.java b/core-cube/src/main/java/org/apache/kylin/cube/model/RowKeyDesc.java
index f1a403d..00557c5 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/RowKeyDesc.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/RowKeyDesc.java
@@ -18,24 +18,23 @@
 
 package org.apache.kylin.cube.model;
 
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Objects;
+import org.apache.commons.lang.ArrayUtils;
+import org.apache.kylin.metadata.model.TblColRef;
+
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 
-import org.apache.commons.lang.ArrayUtils;
-import org.apache.kylin.metadata.model.TblColRef;
-
-import com.fasterxml.jackson.annotation.JsonAutoDetect;
-import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
-import com.fasterxml.jackson.annotation.JsonProperty;
-import com.google.common.base.Objects;
-
 /**
  */
 @JsonAutoDetect(fieldVisibility = Visibility.NONE, getterVisibility = Visibility.NONE, isGetterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE)
-public class RowKeyDesc {
+public class RowKeyDesc implements java.io.Serializable {
 
     @JsonProperty("rowkey_columns")
     private RowKeyColDesc[] rowkeyColumns;

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-cube/src/main/java/org/apache/kylin/cube/model/SelectRule.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/SelectRule.java b/core-cube/src/main/java/org/apache/kylin/cube/model/SelectRule.java
index 63b0fc4..4a6c510 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/SelectRule.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/SelectRule.java
@@ -20,7 +20,7 @@ package org.apache.kylin.cube.model;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 
-public class SelectRule {
+public class SelectRule implements java.io.Serializable {
     @JsonProperty("hierarchy_dims")
     public String[][] hierarchy_dims;
     @JsonProperty("mandatory_dims")

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-dictionary/src/main/java/org/apache/kylin/dict/NumberDictionaryForestBuilder.java
----------------------------------------------------------------------
diff --git a/core-dictionary/src/main/java/org/apache/kylin/dict/NumberDictionaryForestBuilder.java b/core-dictionary/src/main/java/org/apache/kylin/dict/NumberDictionaryForestBuilder.java
index 5502a74..60c9f59 100644
--- a/core-dictionary/src/main/java/org/apache/kylin/dict/NumberDictionaryForestBuilder.java
+++ b/core-dictionary/src/main/java/org/apache/kylin/dict/NumberDictionaryForestBuilder.java
@@ -26,7 +26,7 @@ import org.apache.kylin.dict.NumberDictionary.NumberBytesCodec;
  */
 public class NumberDictionaryForestBuilder extends TrieDictionaryForestBuilder<String> {
 
-    public static class Number2BytesConverter implements BytesConverter<String> {
+    public static class Number2BytesConverter implements BytesConverter<String>, java.io.Serializable {
 
         static final int MAX_DIGITS_BEFORE_DECIMAL_POINT = NumberDictionary.MAX_DIGITS_BEFORE_DECIMAL_POINT;
         static final ThreadLocal<NumberBytesCodec> LOCAL = new ThreadLocal<NumberBytesCodec>();

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-dictionary/src/main/java/org/apache/kylin/dict/StringBytesConverter.java
----------------------------------------------------------------------
diff --git a/core-dictionary/src/main/java/org/apache/kylin/dict/StringBytesConverter.java b/core-dictionary/src/main/java/org/apache/kylin/dict/StringBytesConverter.java
index 0bec6a1..9107a4c 100644
--- a/core-dictionary/src/main/java/org/apache/kylin/dict/StringBytesConverter.java
+++ b/core-dictionary/src/main/java/org/apache/kylin/dict/StringBytesConverter.java
@@ -20,7 +20,7 @@ package org.apache.kylin.dict;
 
 import org.apache.kylin.common.util.Bytes;
 
-public class StringBytesConverter implements BytesConverter<String> {
+public class StringBytesConverter implements BytesConverter<String>, java.io.Serializable {
 
     @Override
     public byte[] convertToBytes(String v) {

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-job/src/main/java/org/apache/kylin/job/execution/ExecutableManager.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/execution/ExecutableManager.java b/core-job/src/main/java/org/apache/kylin/job/execution/ExecutableManager.java
index 0273fd8..466cdad 100644
--- a/core-job/src/main/java/org/apache/kylin/job/execution/ExecutableManager.java
+++ b/core-job/src/main/java/org/apache/kylin/job/execution/ExecutableManager.java
@@ -18,12 +18,9 @@
 
 package org.apache.kylin.job.execution;
 
-import java.lang.reflect.Constructor;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
 import org.apache.kylin.common.KylinConfig;
 import org.apache.kylin.common.util.ClassUtil;
 import org.apache.kylin.job.dao.ExecutableDao;
@@ -34,9 +31,11 @@ import org.apache.kylin.job.exception.PersistentException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
+import java.lang.reflect.Constructor;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 /**
  */
@@ -352,6 +351,10 @@ public class ExecutableManager {
                 }
             }
         }
+
+        if (job.getStatus() == ExecutableState.SUCCEED) {
+            updateJobOutput(job.getId(), ExecutableState.READY, null, null);
+        }
     }
 
     public void pauseJob(String jobId) {

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/dimension/AbstractDateDimEnc.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/dimension/AbstractDateDimEnc.java b/core-metadata/src/main/java/org/apache/kylin/dimension/AbstractDateDimEnc.java
index ec6347f..a54bcda 100644
--- a/core-metadata/src/main/java/org/apache/kylin/dimension/AbstractDateDimEnc.java
+++ b/core-metadata/src/main/java/org/apache/kylin/dimension/AbstractDateDimEnc.java
@@ -18,6 +18,10 @@
 
 package org.apache.kylin.dimension;
 
+import org.apache.kylin.common.util.BytesUtil;
+import org.apache.kylin.common.util.DateFormat;
+import org.apache.kylin.metadata.datatype.DataTypeSerializer;
+
 import java.io.IOException;
 import java.io.ObjectInput;
 import java.io.ObjectOutput;
@@ -25,10 +29,6 @@ import java.io.Serializable;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 
-import org.apache.kylin.common.util.BytesUtil;
-import org.apache.kylin.common.util.DateFormat;
-import org.apache.kylin.metadata.datatype.DataTypeSerializer;
-
 public class AbstractDateDimEnc extends DimensionEncoding {
     private static final long serialVersionUID = 1L;
 
@@ -81,11 +81,9 @@ public class AbstractDateDimEnc extends DimensionEncoding {
     @Override
     public DataTypeSerializer<Object> asDataTypeSerializer() {
         return new DataTypeSerializer<Object>() {
-            // be thread-safe and avoid repeated obj creation
-            private ThreadLocal<byte[]> current = new ThreadLocal<byte[]>();
 
             private byte[] currentBuf() {
-                byte[] buf = current.get();
+                byte[] buf = (byte[]) current.get();
                 if (buf == null) {
                     buf = new byte[fixedLen];
                     current.set(buf);

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/dimension/BooleanDimEnc.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/dimension/BooleanDimEnc.java b/core-metadata/src/main/java/org/apache/kylin/dimension/BooleanDimEnc.java
index c3f4c11..75e50a1 100644
--- a/core-metadata/src/main/java/org/apache/kylin/dimension/BooleanDimEnc.java
+++ b/core-metadata/src/main/java/org/apache/kylin/dimension/BooleanDimEnc.java
@@ -109,11 +109,9 @@ public class BooleanDimEnc extends DimensionEncoding {
     }
 
     private class BooleanSerializer extends DataTypeSerializer<Object> {
-        // be thread-safe and avoid repeated obj creation
-        private ThreadLocal<byte[]> current = new ThreadLocal<byte[]>();
 
         private byte[] currentBuf() {
-            byte[] buf = current.get();
+            byte[] buf = (byte[]) current.get();
             if (buf == null) {
                 buf = new byte[fixedLen];
                 current.set(buf);

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/dimension/DictionaryDimEnc.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/dimension/DictionaryDimEnc.java b/core-metadata/src/main/java/org/apache/kylin/dimension/DictionaryDimEnc.java
index 500b410..48238dc 100644
--- a/core-metadata/src/main/java/org/apache/kylin/dimension/DictionaryDimEnc.java
+++ b/core-metadata/src/main/java/org/apache/kylin/dimension/DictionaryDimEnc.java
@@ -39,12 +39,15 @@ public class DictionaryDimEnc extends DimensionEncoding {
     // ============================================================================
 
     // could use a lazy loading trick here, to prevent loading all dictionaries of a segment at once
-    private final Dictionary<String> dict;
-    private final int fixedLen;
+    private Dictionary<String> dict;
+    private int fixedLen;
 
     // used in encode(), when a value does not exist in dictionary
-    private final int roundingFlag;
-    private final byte defaultByte;
+    private int roundingFlag;
+    private byte defaultByte;
+
+    public DictionaryDimEnc() {
+    }
 
     public DictionaryDimEnc(Dictionary<String> dict) {
         this(dict, 0, NULL);
@@ -145,12 +148,18 @@ public class DictionaryDimEnc extends DimensionEncoding {
 
     @Override
     public void writeExternal(ObjectOutput out) throws IOException {
-        throw new UnsupportedOperationException();
+        out.writeInt(fixedLen);
+        out.writeInt(roundingFlag);
+        out.write(defaultByte);
+        out.writeObject(dict);
     }
 
     @Override
     public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
-        throw new UnsupportedOperationException();
+        this.fixedLen = in.readInt();
+        this.roundingFlag = in.readInt();
+        this.defaultByte = in.readByte();
+        this.dict = (Dictionary<String>) in.readObject();
     }
 
 }

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/dimension/FixedLenDimEnc.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/dimension/FixedLenDimEnc.java b/core-metadata/src/main/java/org/apache/kylin/dimension/FixedLenDimEnc.java
index b219766..f7f02a0 100644
--- a/core-metadata/src/main/java/org/apache/kylin/dimension/FixedLenDimEnc.java
+++ b/core-metadata/src/main/java/org/apache/kylin/dimension/FixedLenDimEnc.java
@@ -129,11 +129,9 @@ public class FixedLenDimEnc extends DimensionEncoding {
     }
 
     public class FixedLenSerializer extends DataTypeSerializer<Object> {
-        // be thread-safe and avoid repeated obj creation
-        private ThreadLocal<byte[]> current = new ThreadLocal<byte[]>();
 
         private byte[] currentBuf() {
-            byte[] buf = current.get();
+            byte[] buf = (byte[]) current.get();
             if (buf == null) {
                 buf = new byte[fixedLen];
                 current.set(buf);

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/dimension/FixedLenHexDimEnc.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/dimension/FixedLenHexDimEnc.java b/core-metadata/src/main/java/org/apache/kylin/dimension/FixedLenHexDimEnc.java
index 83118fc..f90a40e 100644
--- a/core-metadata/src/main/java/org/apache/kylin/dimension/FixedLenHexDimEnc.java
+++ b/core-metadata/src/main/java/org/apache/kylin/dimension/FixedLenHexDimEnc.java
@@ -224,11 +224,9 @@ public class FixedLenHexDimEnc extends DimensionEncoding {
     }
 
     public class FixedLenSerializer extends DataTypeSerializer<Object> {
-        // be thread-safe and avoid repeated obj creation
-        private ThreadLocal<byte[]> current = new ThreadLocal<byte[]>();
 
         private byte[] currentBuf() {
-            byte[] buf = current.get();
+            byte[] buf = (byte[]) current.get();
             if (buf == null) {
                 buf = new byte[bytelen];
                 current.set(buf);

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/dimension/IntDimEnc.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/dimension/IntDimEnc.java b/core-metadata/src/main/java/org/apache/kylin/dimension/IntDimEnc.java
index f25f2a6..aa954da 100644
--- a/core-metadata/src/main/java/org/apache/kylin/dimension/IntDimEnc.java
+++ b/core-metadata/src/main/java/org/apache/kylin/dimension/IntDimEnc.java
@@ -109,11 +109,9 @@ public class IntDimEnc extends DimensionEncoding {
     }
 
     public class IntegerSerializer extends DataTypeSerializer<Object> {
-        // be thread-safe and avoid repeated obj creation
-        private ThreadLocal<byte[]> current = new ThreadLocal<byte[]>();
 
         private byte[] currentBuf() {
-            byte[] buf = current.get();
+            byte[] buf = (byte[]) current.get();
             if (buf == null) {
                 buf = new byte[fixedLen];
                 current.set(buf);

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/dimension/IntegerDimEnc.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/dimension/IntegerDimEnc.java b/core-metadata/src/main/java/org/apache/kylin/dimension/IntegerDimEnc.java
index 090dc83..0875a7f 100644
--- a/core-metadata/src/main/java/org/apache/kylin/dimension/IntegerDimEnc.java
+++ b/core-metadata/src/main/java/org/apache/kylin/dimension/IntegerDimEnc.java
@@ -135,11 +135,9 @@ public class IntegerDimEnc extends DimensionEncoding {
     }
 
     public class IntegerSerializer extends DataTypeSerializer<Object> {
-        // be thread-safe and avoid repeated obj creation
-        private ThreadLocal<byte[]> current = new ThreadLocal<byte[]>();
 
         private byte[] currentBuf() {
-            byte[] buf = current.get();
+            byte[] buf = (byte[]) current.get();
             if (buf == null) {
                 buf = new byte[fixedLen];
                 current.set(buf);

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/dimension/OneMoreByteVLongDimEnc.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/dimension/OneMoreByteVLongDimEnc.java b/core-metadata/src/main/java/org/apache/kylin/dimension/OneMoreByteVLongDimEnc.java
index 993aac3..c0c52d1 100644
--- a/core-metadata/src/main/java/org/apache/kylin/dimension/OneMoreByteVLongDimEnc.java
+++ b/core-metadata/src/main/java/org/apache/kylin/dimension/OneMoreByteVLongDimEnc.java
@@ -18,17 +18,17 @@
 
 package org.apache.kylin.dimension;
 
+import org.apache.kylin.common.util.BytesUtil;
+import org.apache.kylin.metadata.datatype.DataTypeSerializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.io.IOException;
 import java.io.ObjectInput;
 import java.io.ObjectOutput;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 
-import org.apache.kylin.common.util.BytesUtil;
-import org.apache.kylin.metadata.datatype.DataTypeSerializer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 /**
  * not being used yet, prepared for future
  */
@@ -118,11 +118,9 @@ public class OneMoreByteVLongDimEnc extends DimensionEncoding {
     }
 
     public class VLongSerializer extends DataTypeSerializer<Object> {
-        // be thread-safe and avoid repeated obj creation
-        private ThreadLocal<byte[]> current = new ThreadLocal<byte[]>();
 
         private byte[] currentBuf() {
-            byte[] buf = current.get();
+            byte[] buf = (byte[]) current.get();
             if (buf == null) {
                 buf = new byte[byteLen];
                 current.set(buf);

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/measure/BufferedMeasureCodec.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/BufferedMeasureCodec.java b/core-metadata/src/main/java/org/apache/kylin/measure/BufferedMeasureCodec.java
index 8a5481c..44e5708 100644
--- a/core-metadata/src/main/java/org/apache/kylin/measure/BufferedMeasureCodec.java
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/BufferedMeasureCodec.java
@@ -18,26 +18,26 @@
 
 package org.apache.kylin.measure;
 
+import org.apache.kylin.metadata.datatype.DataType;
+import org.apache.kylin.metadata.model.MeasureDesc;
+
 import java.nio.BufferOverflowException;
 import java.nio.ByteBuffer;
 import java.util.Collection;
 
-import org.apache.kylin.metadata.datatype.DataType;
-import org.apache.kylin.metadata.model.MeasureDesc;
-
 /**
  * This class embeds a reusable byte buffer for measure encoding, and is not thread-safe.
  * The buffer will grow to accommodate BufferOverflowException until a limit.
  * The problem here to solve is some measure type cannot provide accurate DataTypeSerializer.maxLength()
  */
 @SuppressWarnings({ "unchecked" })
-public class BufferedMeasureCodec {
+public class BufferedMeasureCodec implements java.io.Serializable {
     public static final int DEFAULT_BUFFER_SIZE = 1024 * 1024; // 1 MB
     public static final int MAX_BUFFER_SIZE = 1 * 1024 * DEFAULT_BUFFER_SIZE; // 1 GB
 
     final private MeasureCodec codec;
 
-    private ByteBuffer buf;
+    private transient ByteBuffer buf;
     final private int[] measureSizes;
 
     public BufferedMeasureCodec(Collection<MeasureDesc> measureDescs) {

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/measure/MeasureCodec.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/MeasureCodec.java b/core-metadata/src/main/java/org/apache/kylin/measure/MeasureCodec.java
index edaf806..2d73e59 100644
--- a/core-metadata/src/main/java/org/apache/kylin/measure/MeasureCodec.java
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/MeasureCodec.java
@@ -18,19 +18,19 @@
 
 package org.apache.kylin.measure;
 
-import java.nio.ByteBuffer;
-import java.util.Collection;
-
 import org.apache.kylin.metadata.datatype.DataType;
 import org.apache.kylin.metadata.datatype.DataTypeSerializer;
 import org.apache.kylin.metadata.model.MeasureDesc;
 
+import java.nio.ByteBuffer;
+import java.util.Collection;
+
 /**
  * @author yangli9
  * 
  */
 @SuppressWarnings({ "rawtypes" })
-public class MeasureCodec {
+public class MeasureCodec implements java.io.Serializable {
 
     private int nMeasures;
     private DataTypeSerializer[] serializers;

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/measure/MeasureIngester.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/MeasureIngester.java b/core-metadata/src/main/java/org/apache/kylin/measure/MeasureIngester.java
index 0076252..26b7298 100644
--- a/core-metadata/src/main/java/org/apache/kylin/measure/MeasureIngester.java
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/MeasureIngester.java
@@ -18,14 +18,14 @@
 
 package org.apache.kylin.measure;
 
-import java.util.Collection;
-import java.util.Map;
-
 import org.apache.kylin.common.util.Dictionary;
 import org.apache.kylin.metadata.model.MeasureDesc;
 import org.apache.kylin.metadata.model.TblColRef;
 
-abstract public class MeasureIngester<V> {
+import java.util.Collection;
+import java.util.Map;
+
+abstract public class MeasureIngester<V> implements java.io.Serializable {
 
     public static MeasureIngester<?> create(MeasureDesc measure) {
         return measure.getFunction().getMeasureType().newIngester();
@@ -42,6 +42,10 @@ abstract public class MeasureIngester<V> {
 
     abstract public V valueOf(String[] values, MeasureDesc measureDesc, Map<TblColRef, Dictionary<String>> dictionaryMap);
 
+    public void reset() {
+
+    }
+
     public V reEncodeDictionary(V value, MeasureDesc measureDesc, Map<TblColRef, Dictionary<String>> oldDicts, Map<TblColRef, Dictionary<String>> newDicts) {
         throw new UnsupportedOperationException();
     }

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/measure/MeasureType.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/MeasureType.java b/core-metadata/src/main/java/org/apache/kylin/measure/MeasureType.java
index 89ff382..3338c8c 100644
--- a/core-metadata/src/main/java/org/apache/kylin/measure/MeasureType.java
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/MeasureType.java
@@ -18,11 +18,6 @@
 
 package org.apache.kylin.measure;
 
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-
 import org.apache.kylin.common.util.Dictionary;
 import org.apache.kylin.metadata.model.FunctionDesc;
 import org.apache.kylin.metadata.model.MeasureDesc;
@@ -32,13 +27,18 @@ import org.apache.kylin.metadata.realization.SQLDigest;
 import org.apache.kylin.metadata.tuple.Tuple;
 import org.apache.kylin.metadata.tuple.TupleInfo;
 
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
 /**
  * MeasureType captures how a kind of aggregation is defined, how it is calculated 
  * during cube build, and how it is involved in query and storage scan.
  * 
  * @param <T> the Java type of aggregation data object, e.g. HLLCounter
  */
-abstract public class MeasureType<T> {
+abstract public class MeasureType<T> implements java.io.Serializable {
 
     /* ============================================================================
      * Define

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/measure/basic/DoubleIngester.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/basic/DoubleIngester.java b/core-metadata/src/main/java/org/apache/kylin/measure/basic/DoubleIngester.java
index a2f3980..e42f275 100644
--- a/core-metadata/src/main/java/org/apache/kylin/measure/basic/DoubleIngester.java
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/basic/DoubleIngester.java
@@ -43,4 +43,9 @@ public class DoubleIngester extends MeasureIngester<DoubleMutable> {
             l.set(Double.parseDouble(values[0]));
         return l;
     }
+
+    @Override
+    public void reset() {
+        current = new DoubleMutable();
+    }
 }

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/measure/basic/DoubleSumAggregator.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/basic/DoubleSumAggregator.java b/core-metadata/src/main/java/org/apache/kylin/measure/basic/DoubleSumAggregator.java
index f276817..29eb787 100644
--- a/core-metadata/src/main/java/org/apache/kylin/measure/basic/DoubleSumAggregator.java
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/basic/DoubleSumAggregator.java
@@ -30,7 +30,8 @@ public class DoubleSumAggregator extends MeasureAggregator<DoubleMutable> {
 
     @Override
     public void reset() {
-        sum.set(0.0);
+//        sum.set(0.0);
+        sum = new DoubleMutable();
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/measure/basic/LongIngester.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/basic/LongIngester.java b/core-metadata/src/main/java/org/apache/kylin/measure/basic/LongIngester.java
index 45a1634..439f096 100644
--- a/core-metadata/src/main/java/org/apache/kylin/measure/basic/LongIngester.java
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/basic/LongIngester.java
@@ -43,4 +43,9 @@ public class LongIngester extends MeasureIngester<LongMutable> {
             l.set(Long.parseLong(values[0]));
         return l;
     }
+
+    @Override
+    public void reset() {
+        current = new LongMutable();
+    }
 }

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/measure/basic/LongSumAggregator.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/basic/LongSumAggregator.java b/core-metadata/src/main/java/org/apache/kylin/measure/basic/LongSumAggregator.java
index e7fdc9d..0a56af8 100644
--- a/core-metadata/src/main/java/org/apache/kylin/measure/basic/LongSumAggregator.java
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/basic/LongSumAggregator.java
@@ -30,7 +30,8 @@ public class LongSumAggregator extends MeasureAggregator<LongMutable> {
 
     @Override
     public void reset() {
-        sum.set(0);
+//        sum.set(0);
+        sum = new LongMutable();
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/measure/bitmap/BitmapCounter.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/bitmap/BitmapCounter.java b/core-metadata/src/main/java/org/apache/kylin/measure/bitmap/BitmapCounter.java
index 827390d..caab094 100644
--- a/core-metadata/src/main/java/org/apache/kylin/measure/bitmap/BitmapCounter.java
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/bitmap/BitmapCounter.java
@@ -18,6 +18,9 @@
 
 package org.apache.kylin.measure.bitmap;
 
+import org.apache.commons.io.IOUtils;
+import org.roaringbitmap.buffer.MutableRoaringBitmap;
+
 import java.io.ByteArrayOutputStream;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
@@ -26,13 +29,10 @@ import java.io.InputStream;
 import java.nio.ByteBuffer;
 import java.util.Iterator;
 
-import org.apache.commons.io.IOUtils;
-import org.roaringbitmap.buffer.MutableRoaringBitmap;
-
 /**
  * Created by sunyerui on 15/12/1.
  */
-public class BitmapCounter implements Comparable<BitmapCounter> {
+public class BitmapCounter implements Comparable<BitmapCounter>, java.io.Serializable {
 
     private MutableRoaringBitmap bitmap = new MutableRoaringBitmap();
 

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/measure/bitmap/BitmapMeasureType.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/bitmap/BitmapMeasureType.java b/core-metadata/src/main/java/org/apache/kylin/measure/bitmap/BitmapMeasureType.java
index 8e2b2f7..6ad82a1 100644
--- a/core-metadata/src/main/java/org/apache/kylin/measure/bitmap/BitmapMeasureType.java
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/bitmap/BitmapMeasureType.java
@@ -138,6 +138,11 @@ public class BitmapMeasureType extends MeasureType<BitmapCounter> {
                 }
                 return retValue;
             }
+
+            @Override
+            public void reset() {
+                current = new BitmapCounter();
+            }
         };
     }
 

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/measure/hllc/HLLCMeasureType.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/hllc/HLLCMeasureType.java b/core-metadata/src/main/java/org/apache/kylin/measure/hllc/HLLCMeasureType.java
index 9601653..b2d2ddc 100644
--- a/core-metadata/src/main/java/org/apache/kylin/measure/hllc/HLLCMeasureType.java
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/hllc/HLLCMeasureType.java
@@ -105,6 +105,11 @@ public class HLLCMeasureType extends MeasureType<HLLCounter> {
                 }
                 return hllc;
             }
+
+            @Override
+            public void reset() {
+                current = new HyperLogLogPlusCounter(dataType.getPrecision());
+            }
         };
     }
 

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/measure/hllc/HLLCSerializer.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/hllc/HLLCSerializer.java b/core-metadata/src/main/java/org/apache/kylin/measure/hllc/HLLCSerializer.java
index e0992c7..df0cfaf 100644
--- a/core-metadata/src/main/java/org/apache/kylin/measure/hllc/HLLCSerializer.java
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/hllc/HLLCSerializer.java
@@ -18,21 +18,18 @@
 
 package org.apache.kylin.measure.hllc;
 
-import java.io.IOException;
-import java.nio.ByteBuffer;
-
 import org.apache.kylin.metadata.datatype.DataType;
 import org.apache.kylin.metadata.datatype.DataTypeSerializer;
 
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
 /**
  * @author yangli9
  * 
  */
 public class HLLCSerializer extends DataTypeSerializer<HLLCounter> {
 
-    // be thread-safe and avoid repeated obj creation
-    private ThreadLocal<HLLCounter> current = new ThreadLocal<HLLCounter>();
-
     private int precision;
 
     public HLLCSerializer(DataType type) {
@@ -49,7 +46,7 @@ public class HLLCSerializer extends DataTypeSerializer<HLLCounter> {
     }
 
     private HLLCounter current() {
-        HLLCounter hllc = current.get();
+        HLLCounter hllc = (HLLCounter) current.get();
         if (hllc == null) {
             hllc = new HLLCounter(precision);
             current.set(hllc);

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/measure/raw/RawSerializer.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/raw/RawSerializer.java b/core-metadata/src/main/java/org/apache/kylin/measure/raw/RawSerializer.java
index 021c146..68a0273 100644
--- a/core-metadata/src/main/java/org/apache/kylin/measure/raw/RawSerializer.java
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/raw/RawSerializer.java
@@ -34,13 +34,11 @@ public class RawSerializer extends DataTypeSerializer<List<ByteArray>> {
     //FIXME to config this and RowConstants.ROWVALUE_BUFFER_SIZE in properties file
     public static final int RAW_BUFFER_SIZE = 1024 * 1024;//1M
 
-    private ThreadLocal<List<ByteArray>> current = new ThreadLocal<>();
-
     public RawSerializer(DataType dataType) {
     }
 
     private List<ByteArray> current() {
-        List<ByteArray> l = current.get();
+        List<ByteArray> l = (List<ByteArray>) current.get();
         if (l == null) {
             l = new ArrayList<ByteArray>();
             current.set(l);

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/measure/topn/TopNCounter.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/topn/TopNCounter.java b/core-metadata/src/main/java/org/apache/kylin/measure/topn/TopNCounter.java
index caf7961..5e4b91e 100644
--- a/core-metadata/src/main/java/org/apache/kylin/measure/topn/TopNCounter.java
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/topn/TopNCounter.java
@@ -40,7 +40,7 @@ import com.google.common.collect.Maps;
  *
  * @param <T> type of data in the stream to be summarized
  */
-public class TopNCounter<T> implements Iterable<Counter<T>> {
+public class TopNCounter<T> implements Iterable<Counter<T>>, java.io.Serializable {
 
     public static final int EXTRA_SPACE_RATE = 50;
 

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/measure/topn/TopNMeasureType.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/topn/TopNMeasureType.java b/core-metadata/src/main/java/org/apache/kylin/measure/topn/TopNMeasureType.java
index c29af6c..8c8b5a6 100644
--- a/core-metadata/src/main/java/org/apache/kylin/measure/topn/TopNMeasureType.java
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/topn/TopNMeasureType.java
@@ -24,6 +24,7 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 
+import com.google.common.collect.Lists;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.kylin.common.util.ByteArray;
 import org.apache.kylin.common.util.Dictionary;
@@ -46,8 +47,6 @@ import org.apache.kylin.metadata.tuple.TupleInfo;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.google.common.collect.Lists;
-
 public class TopNMeasureType extends MeasureType<TopNCounter<ByteArray>> {
 
     private static final Logger logger = LoggerFactory.getLogger(TopNMeasureType.class);
@@ -156,6 +155,7 @@ public class TopNMeasureType extends MeasureType<TopNCounter<ByteArray>> {
                 return topNCounter;
             }
 
+
             @Override
             public TopNCounter<ByteArray> reEncodeDictionary(TopNCounter<ByteArray> value, MeasureDesc measureDesc, Map<TblColRef, Dictionary<String>> oldDicts, Map<TblColRef, Dictionary<String>> newDicts) {
                 TopNCounter<ByteArray> topNCounter = value;

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/BigDecimalSerializer.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/BigDecimalSerializer.java b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/BigDecimalSerializer.java
index 64968b8..b5043f5 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/BigDecimalSerializer.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/BigDecimalSerializer.java
@@ -35,7 +35,7 @@ public class BigDecimalSerializer extends DataTypeSerializer<BigDecimal> {
     private static final Logger logger = LoggerFactory.getLogger(BigDecimalSerializer.class);
 
     final DataType type;
-    transient final int maxLength;
+    final int maxLength;
 
     transient int avoidVerbose = 0;
 

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/BooleanSerializer.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/BooleanSerializer.java b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/BooleanSerializer.java
index acb6de1..998e5e9 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/BooleanSerializer.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/BooleanSerializer.java
@@ -18,18 +18,15 @@
 
 package org.apache.kylin.metadata.datatype;
 
-import java.nio.ByteBuffer;
-
 import org.apache.commons.lang.ArrayUtils;
 import org.apache.commons.lang.BooleanUtils;
 
+import java.nio.ByteBuffer;
+
 public class BooleanSerializer extends DataTypeSerializer<LongMutable> {
 
     public final static String[] TRUE_VALUE_SET = { "true", "t", "on", "yes" };
 
-    // be thread-safe and avoid repeated obj creation
-    private ThreadLocal<LongMutable> current = new ThreadLocal<LongMutable>();
-
     public BooleanSerializer(DataType type) {
     }
 
@@ -39,7 +36,7 @@ public class BooleanSerializer extends DataTypeSerializer<LongMutable> {
     }
 
     private LongMutable current() {
-        LongMutable l = current.get();
+        LongMutable l = (LongMutable) current.get();
         if (l == null) {
             l = new LongMutable();
             current.set(l);

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DataTypeSerializer.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DataTypeSerializer.java b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DataTypeSerializer.java
index a739377..a4a35a4 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DataTypeSerializer.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DataTypeSerializer.java
@@ -18,19 +18,21 @@
 
 package org.apache.kylin.metadata.datatype;
 
-import java.nio.ByteBuffer;
-import java.util.Map;
-
+import com.google.common.collect.Maps;
 import org.apache.kylin.common.util.BytesSerializer;
 
-import com.google.common.collect.Maps;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.nio.ByteBuffer;
+import java.util.Map;
 
 /**
  * Note: the implementations MUST be thread-safe.
  */
-abstract public class DataTypeSerializer<T> implements BytesSerializer<T> {
+abstract public class DataTypeSerializer<T> implements BytesSerializer<T>, java.io.Serializable {
 
     final static Map<String, Class<?>> implementations = Maps.newHashMap();
+    protected transient ThreadLocal current = new ThreadLocal();
     static {
         implementations.put("char", StringSerializer.class);
         implementations.put("varchar", StringSerializer.class);
@@ -94,4 +96,9 @@ abstract public class DataTypeSerializer<T> implements BytesSerializer<T> {
         else
             return value.toString();
     }
+
+    private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+        in.defaultReadObject();
+        current = new ThreadLocal();
+    }
 }

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DateTimeSerializer.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DateTimeSerializer.java b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DateTimeSerializer.java
index 07f98b3..a5719bd 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DateTimeSerializer.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DateTimeSerializer.java
@@ -18,14 +18,11 @@
 
 package org.apache.kylin.metadata.datatype;
 
-import java.nio.ByteBuffer;
-
 import org.apache.kylin.common.util.DateFormat;
 
-public class DateTimeSerializer extends DataTypeSerializer<LongMutable> {
+import java.nio.ByteBuffer;
 
-    // be thread-safe and avoid repeated obj creation
-    private ThreadLocal<LongMutable> current = new ThreadLocal<LongMutable>();
+public class DateTimeSerializer extends DataTypeSerializer<LongMutable> {
 
     public DateTimeSerializer(DataType type) {
     }
@@ -36,7 +33,7 @@ public class DateTimeSerializer extends DataTypeSerializer<LongMutable> {
     }
 
     private LongMutable current() {
-        LongMutable l = current.get();
+        LongMutable l = (LongMutable) current.get();
         if (l == null) {
             l = new LongMutable();
             current.set(l);

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DoubleSerializer.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DoubleSerializer.java b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DoubleSerializer.java
index 976dc51..cda4ff5 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DoubleSerializer.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/DoubleSerializer.java
@@ -24,9 +24,6 @@ import java.nio.ByteBuffer;
  */
 public class DoubleSerializer extends DataTypeSerializer<DoubleMutable> {
 
-    // be thread-safe and avoid repeated obj creation
-    private ThreadLocal<DoubleMutable> current = new ThreadLocal<DoubleMutable>();
-
     public DoubleSerializer(DataType type) {
     }
 
@@ -36,7 +33,7 @@ public class DoubleSerializer extends DataTypeSerializer<DoubleMutable> {
     }
 
     private DoubleMutable current() {
-        DoubleMutable d = current.get();
+        DoubleMutable d = (DoubleMutable) current.get();
         if (d == null) {
             d = new DoubleMutable();
             current.set(d);

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/Int4Serializer.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/Int4Serializer.java b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/Int4Serializer.java
index 7b95505..c726ab6 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/Int4Serializer.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/Int4Serializer.java
@@ -18,17 +18,14 @@
 
 package org.apache.kylin.metadata.datatype;
 
-import java.nio.ByteBuffer;
-
 import org.apache.kylin.common.util.BytesUtil;
 
+import java.nio.ByteBuffer;
+
 /**
  */
 public class Int4Serializer extends DataTypeSerializer<IntMutable> {
 
-    // be thread-safe and avoid repeated obj creation
-    private ThreadLocal<IntMutable> current = new ThreadLocal<IntMutable>();
-
     public Int4Serializer(DataType type) {
     }
 
@@ -38,7 +35,7 @@ public class Int4Serializer extends DataTypeSerializer<IntMutable> {
     }
 
     private IntMutable current() {
-        IntMutable l = current.get();
+        IntMutable l = (IntMutable) current.get();
         if (l == null) {
             l = new IntMutable();
             current.set(l);

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/Long8Serializer.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/Long8Serializer.java b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/Long8Serializer.java
index fa333b2..186cdb9 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/Long8Serializer.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/Long8Serializer.java
@@ -18,17 +18,14 @@
 
 package org.apache.kylin.metadata.datatype;
 
-import java.nio.ByteBuffer;
-
 import org.apache.kylin.common.util.BytesUtil;
 
+import java.nio.ByteBuffer;
+
 /**
  */
 public class Long8Serializer extends DataTypeSerializer<LongMutable> {
 
-    // be thread-safe and avoid repeated obj creation
-    private ThreadLocal<LongMutable> current = new ThreadLocal<LongMutable>();
-
     public Long8Serializer(DataType type) {
     }
 
@@ -38,7 +35,7 @@ public class Long8Serializer extends DataTypeSerializer<LongMutable> {
     }
 
     private LongMutable current() {
-        LongMutable l = current.get();
+        LongMutable l = (LongMutable) current.get();
         if (l == null) {
             l = new LongMutable();
             current.set(l);

http://git-wip-us.apache.org/repos/asf/kylin/blob/78e6cd5b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/LongSerializer.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/LongSerializer.java b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/LongSerializer.java
index 9306a70..d8f3f37 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/LongSerializer.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/datatype/LongSerializer.java
@@ -18,17 +18,14 @@
 
 package org.apache.kylin.metadata.datatype;
 
-import java.nio.ByteBuffer;
-
 import org.apache.kylin.common.util.BytesUtil;
 
+import java.nio.ByteBuffer;
+
 /**
  */
 public class LongSerializer extends DataTypeSerializer<LongMutable> {
 
-    // be thread-safe and avoid repeated obj creation
-    private ThreadLocal<LongMutable> current = new ThreadLocal<LongMutable>();
-
     public LongSerializer(DataType type) {
     }
 
@@ -38,7 +35,7 @@ public class LongSerializer extends DataTypeSerializer<LongMutable> {
     }
 
     private LongMutable current() {
-        LongMutable l = current.get();
+        LongMutable l = (LongMutable) current.get();
         if (l == null) {
             l = new LongMutable();
             current.set(l);


Mime
View raw message