carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ravipes...@apache.org
Subject [31/47] incubator-carbondata git commit: [CARBONDATA-111]Need to catch exception in compaction and rethrow it to spark. (#858)
Date Mon, 01 Aug 2016 10:05:29 GMT
[CARBONDATA-111]Need to catch exception in compaction  and rethrow it to spark. (#858)

 Need to catch exception and rethrow it to spark. need to unlock the compaction lock in case
of exception.
 Need to set the property spark.job.interruptOnCancel should be set to true so that if the
user kills the job in UI.
The task will also be killed. if any exception occurs in compaction then clean up the HDFS
folders.


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

Branch: refs/heads/master
Commit: 973def99f72c267e625497daf2ece18c18767d73
Parents: bdc1321
Author: ravikiran23 <ravikiran.sn042@gmail.com>
Authored: Thu Jul 28 21:03:20 2016 +0530
Committer: sujith71955 <sujithchacko.2010@gmail.com>
Committed: Thu Jul 28 21:03:20 2016 +0530

----------------------------------------------------------------------
 .../spark/rdd/CarbonDataRDDFactory.scala        |  88 ++++++------
 .../carbondata/spark/rdd/CarbonMergerRDD.scala  | 136 ++++++++++---------
 2 files changed, 122 insertions(+), 102 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/973def99/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDataRDDFactory.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDataRDDFactory.scala
b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDataRDDFactory.scala
index 15de7bf..54cf1bd 100644
--- a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDataRDDFactory.scala
+++ b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDataRDDFactory.scala
@@ -400,53 +400,44 @@ object CarbonDataRDDFactory extends Logging {
       new Thread {
         override def run(): Unit = {
 
-          while (loadsToMerge.size() > 1) {
-          // Deleting the any partially loaded data if present.
-          // in some case the segment folder which is present in store will not have entry
in
-          // status.
-          // so deleting those folders.
           try {
-            CarbonLoaderUtil.deletePartialLoadDataIfExist(carbonLoadModel, true)
-          }
-          catch {
-            case e: Exception =>
-              logger
-                .error("Exception in compaction thread while clean up of stale segments "
+ e
-                  .getMessage
-                )
-          }
-            val futureList: util.List[Future[Void]] = new util.ArrayList[Future[Void]](
-              CarbonCommonConstants
-                .DEFAULT_COLLECTION_SIZE
-            )
+            while (loadsToMerge.size() > 1) {
+              deletePartialLoadsInCompaction(carbonLoadModel)
+              val futureList: util.List[Future[Void]] = new util.ArrayList[Future[Void]](
+                CarbonCommonConstants
+                  .DEFAULT_COLLECTION_SIZE
+              )
 
-            scanSegmentsAndSubmitJob(futureList)
+              scanSegmentsAndSubmitJob(futureList)
 
-            futureList.asScala.foreach(future => {
-              try {
+              futureList.asScala.foreach(future => {
                 future.get
               }
-              catch {
-                case e: Exception =>
-                  logger.error("Exception in compaction thread " + e.getMessage)
-              }
+              )
+              // scan again and deterrmine if anything is there to merge again.
+              readLoadMetadataDetails(carbonLoadModel, hdfsStoreLocation)
+              segList = carbonLoadModel.getLoadMetadataDetails
+
+              loadsToMerge = CarbonDataMergerUtil.identifySegmentsToBeMerged(
+                hdfsStoreLocation,
+                carbonLoadModel,
+                partitioner.partitionCount,
+                compactionModel.compactionSize,
+                segList,
+                compactionModel.compactionType
+              )
             }
-            )
-            // scan again and deterrmine if anything is there to merge again.
-            readLoadMetadataDetails(carbonLoadModel, hdfsStoreLocation)
-            segList = carbonLoadModel.getLoadMetadataDetails
-
-            loadsToMerge = CarbonDataMergerUtil.identifySegmentsToBeMerged(
-              hdfsStoreLocation,
-              carbonLoadModel,
-              partitioner.partitionCount,
-              compactionModel.compactionSize,
-              segList,
-              compactionModel.compactionType
-            )
           }
-          executor.shutdown()
-          compactionLock.unlock()
+          catch {
+            case e: Exception =>
+              logger.error("Exception in compaction thread " + e.getMessage)
+              throw e
+          }
+          finally {
+            executor.shutdownNow()
+            deletePartialLoadsInCompaction(carbonLoadModel)
+            compactionLock.unlock()
+          }
         }
       }.start
     }
@@ -504,6 +495,23 @@ object CarbonDataRDDFactory extends Logging {
     }
   }
 
+  def deletePartialLoadsInCompaction(carbonLoadModel: CarbonLoadModel): Unit = {
+    // Deleting the any partially loaded data if present.
+    // in some case the segment folder which is present in store will not have entry in
+    // status.
+    // so deleting those folders.
+    try {
+      CarbonLoaderUtil.deletePartialLoadDataIfExist(carbonLoadModel, true)
+    }
+    catch {
+      case e: Exception =>
+        logger
+          .error("Exception in compaction thread while clean up of stale segments " + e
+            .getMessage
+          )
+    }
+  }
+
   def loadCarbonData(sc: SQLContext,
       carbonLoadModel: CarbonLoadModel,
       storeLocation: String,

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/973def99/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonMergerRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonMergerRDD.scala
b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonMergerRDD.scala
index 0129a2c..254d51b 100644
--- a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonMergerRDD.scala
+++ b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonMergerRDD.scala
@@ -57,6 +57,7 @@ class CarbonMergerRDD[K, V](
 
   val defaultParallelism = sc.defaultParallelism
   sc.setLocalProperty("spark.scheduler.pool", "DDL")
+  sc.setLocalProperty("spark.job.interruptOnCancel", "true")
 
   val storeLocation = carbonMergerMapping.storeLocation
   val hdfsStoreLocation = carbonMergerMapping.hdfsStoreLocation
@@ -68,83 +69,94 @@ class CarbonMergerRDD[K, V](
   override def compute(theSplit: Partition, context: TaskContext): Iterator[(K, V)] = {
     val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
     val iter = new Iterator[(K, V)] {
-      var dataloadStatus = CarbonCommonConstants.STORE_LOADSTATUS_FAILURE
-      carbonLoadModel.setTaskNo(String.valueOf(theSplit.index))
-      val carbonSparkPartition = theSplit.asInstanceOf[CarbonSparkPartition]
 
-      val tempLocationKey: String = carbonLoadModel.getDatabaseName + '_' + carbonLoadModel
-        .getTableName + carbonLoadModel.getTaskNo
-      CarbonProperties.getInstance().addProperty(tempLocationKey, storeLocation)
+      var mergeStatus = false
+      try {
+        var dataloadStatus = CarbonCommonConstants.STORE_LOADSTATUS_FAILURE
+        carbonLoadModel.setTaskNo(String.valueOf(theSplit.index))
+        val carbonSparkPartition = theSplit.asInstanceOf[CarbonSparkPartition]
 
-      // sorting the table block info List.
-      var tableBlockInfoList = carbonSparkPartition.tableBlockInfos
+        val tempLocationKey: String = carbonLoadModel.getDatabaseName + '_' + carbonLoadModel
+          .getTableName + carbonLoadModel.getTaskNo
+        CarbonProperties.getInstance().addProperty(tempLocationKey, storeLocation)
 
-      Collections.sort(tableBlockInfoList)
+        // sorting the table block info List.
+        var tableBlockInfoList = carbonSparkPartition.tableBlockInfos
 
-      val segmentMapping: java.util.Map[String, TaskBlockInfo] =
-        CarbonCompactionUtil.createMappingForSegments(tableBlockInfoList)
+        Collections.sort(tableBlockInfoList)
 
-      val dataFileMetadataSegMapping: java.util.Map[String, List[DataFileFooter]] =
-        CarbonCompactionUtil.createDataFileFooterMappingForSegments(tableBlockInfoList)
+        val segmentMapping: java.util.Map[String, TaskBlockInfo] =
+          CarbonCompactionUtil.createMappingForSegments(tableBlockInfoList)
 
-      carbonLoadModel.setStorePath(hdfsStoreLocation)
+        val dataFileMetadataSegMapping: java.util.Map[String, List[DataFileFooter]] =
+          CarbonCompactionUtil.createDataFileFooterMappingForSegments(tableBlockInfoList)
 
-      // taking the last table block info for getting the segment properties.
-      val listMetadata = dataFileMetadataSegMapping.get(tableBlockInfoList.get
-      (tableBlockInfoList.size()-1).getSegmentId())
+        carbonLoadModel.setStorePath(hdfsStoreLocation)
 
-      val colCardinality: Array[Int] = listMetadata.get(listMetadata.size() - 1).getSegmentInfo
-        .getColumnCardinality
+        // taking the last table block info for getting the segment properties.
+        val listMetadata = dataFileMetadataSegMapping.get(tableBlockInfoList.get
+        (tableBlockInfoList.size() - 1).getSegmentId()
+        )
 
-      val segmentProperties = new SegmentProperties(
-        listMetadata.get(listMetadata.size() - 1).getColumnInTable,
-        colCardinality
-      )
+        val colCardinality: Array[Int] = listMetadata.get(listMetadata.size() - 1).getSegmentInfo
+          .getColumnCardinality
 
-      val exec = new CarbonCompactionExecutor(segmentMapping, segmentProperties, schemaName,
-        factTableName, hdfsStoreLocation, carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable,
-        dataFileMetadataSegMapping
-      )
+        val segmentProperties = new SegmentProperties(
+          listMetadata.get(listMetadata.size() - 1).getColumnInTable,
+          colCardinality
+        )
 
-      // fire a query and get the results.
-      var result2: util.List[RawResultIterator] = null
-      try {
-        result2 = exec.processTableBlocks()
-      } catch {
-        case e: Throwable =>
-          exec.clearDictionaryFromQueryModel
-          LOGGER.error(e)
-          if (null != e.getMessage) {
-            sys.error("Exception occurred in query execution :: " + e.getMessage)
-          } else {
-            sys.error("Exception occurred in query execution.Please check logs.")
-          }
-      }
+        val exec = new CarbonCompactionExecutor(segmentMapping, segmentProperties, schemaName,
+          factTableName, hdfsStoreLocation, carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable,
+          dataFileMetadataSegMapping
+        )
 
-      val mergeNumber = mergedLoadName
-        .substring(mergedLoadName.lastIndexOf(CarbonCommonConstants.LOAD_FOLDER) +
-          CarbonCommonConstants.LOAD_FOLDER.length(), mergedLoadName.length()
+        // fire a query and get the results.
+        var result2: util.List[RawResultIterator] = null
+        try {
+          result2 = exec.processTableBlocks()
+        } catch {
+          case e: Throwable =>
+            exec.clearDictionaryFromQueryModel
+            LOGGER.error(e)
+            if (null != e.getMessage) {
+              sys.error("Exception occurred in query execution :: " + e.getMessage)
+            } else {
+              sys.error("Exception occurred in query execution.Please check logs.")
+            }
+        }
+
+        val mergeNumber = mergedLoadName
+          .substring(mergedLoadName.lastIndexOf(CarbonCommonConstants.LOAD_FOLDER) +
+            CarbonCommonConstants.LOAD_FOLDER.length(), mergedLoadName.length()
+          )
+
+        val tempStoreLoc = CarbonDataProcessorUtil.getLocalDataFolderLocation(schemaName,
+          factTableName,
+          carbonLoadModel.getTaskNo,
+          "0",
+          mergeNumber
         )
 
-      val tempStoreLoc = CarbonDataProcessorUtil.getLocalDataFolderLocation(schemaName,
-        factTableName,
-        carbonLoadModel.getTaskNo,
-        "0",
-        mergeNumber
-      )
+        carbonLoadModel.setSegmentId(mergeNumber)
+        carbonLoadModel.setPartitionId("0")
+        val merger =
+          new RowResultMerger(result2,
+            schemaName,
+            factTableName,
+            segmentProperties,
+            tempStoreLoc,
+            carbonLoadModel,
+            colCardinality
+          )
+        mergeStatus = merger.mergerSlice()
 
-      carbonLoadModel.setSegmentId(mergeNumber)
-      carbonLoadModel.setPartitionId("0")
-      val merger =
-        new RowResultMerger(result2,
-        schemaName,
-        factTableName,
-        segmentProperties,
-        tempStoreLoc,
-        carbonLoadModel,
-        colCardinality
-      )
-      val mergeStatus = merger.mergerSlice()
+      }
+      catch {
+        case e: Exception =>
+          LOGGER.error(e)
+          throw e
+      }
 
       var finished = false
 


Mime
View raw message