carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jack...@apache.org
Subject [1/2] incubator-carbondata git commit: fix bug for reading dataframe concurrently
Date Fri, 02 Dec 2016 13:44:31 GMT
Repository: incubator-carbondata
Updated Branches:
  refs/heads/master e7e370cac -> 9dd09659a


fix bug for reading dataframe concurrently


Project: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/commit/9dcdf7de
Tree: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/tree/9dcdf7de
Diff: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/diff/9dcdf7de

Branch: refs/heads/master
Commit: 9dcdf7de6bde64d1c800fd268f2099d2278e8f33
Parents: e7e370c
Author: QiangCai <qiangcai@qq.com>
Authored: Fri Dec 2 17:41:23 2016 +0800
Committer: QiangCai <qiangcai@qq.com>
Committed: Fri Dec 2 17:46:56 2016 +0800

----------------------------------------------------------------------
 .../carbondata/spark/rdd/CarbonDataLoadRDD.scala     | 15 ++++++++++++++-
 1 file changed, 14 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/9dcdf7de/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataLoadRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataLoadRDD.scala
b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataLoadRDD.scala
index 319d85c..5d6a663 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataLoadRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataLoadRDD.scala
@@ -18,6 +18,7 @@
 package org.apache.carbondata.spark.rdd
 
 import java.lang.Long
+import java.nio.ByteBuffer
 import java.text.SimpleDateFormat
 import java.util
 import java.util.UUID
@@ -548,10 +549,22 @@ class DataFrameLoaderRDD[K, V](
 class PartitionIterator(partitionIter: Iterator[DataLoadPartitionWrap[Row]],
     carbonLoadModel: CarbonLoadModel,
     context: TaskContext) extends JavaRddIterator[JavaRddIterator[Array[String]]] {
+  val serializer = SparkEnv.get.closureSerializer.newInstance()
+  var serializeBuffer: ByteBuffer = null
   def hasNext: Boolean = partitionIter.hasNext
+
   def next: JavaRddIterator[Array[String]] = {
     val value = partitionIter.next
-    new RddIterator(value.rdd.iterator(value.partition, context),
+    // The rdd (which come from Hive Table) don't support to read dataframe concurrently.
+    // So here will create different rdd instance for each thread.
+    val newInstance = {
+      if (serializeBuffer == null) {
+        serializeBuffer = serializer.serialize[RDD[Row]](value.rdd)
+      }
+      serializeBuffer.rewind()
+      serializer.deserialize[RDD[Row]](serializeBuffer)
+    }
+    new RddIterator(newInstance.iterator(value.partition, context),
         carbonLoadModel,
         context)
   }


Mime
View raw message