carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From akash...@apache.org
Subject [carbondata] branch master updated: [CARBONDATA-3852] [CARBONDATA-3851] Fix Data update issue with SCD/CDC for partition table On spark cluster deploy mode
Date Mon, 13 Jul 2020 17:59:39 GMT
This is an automated email from the ASF dual-hosted git repository.

akashrn5 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/carbondata.git


The following commit(s) were added to refs/heads/master by this push:
     new ea253a1  [CARBONDATA-3852] [CARBONDATA-3851] Fix Data update issue with SCD/CDC for
partition table On spark cluster deploy mode
ea253a1 is described below

commit ea253a11836840f207d0795e8ccaada4e1fd10db
Author: Indhumathi27 <indhumathim27@gmail.com>
AuthorDate: Fri Jul 10 22:58:02 2020 +0530

    [CARBONDATA-3852] [CARBONDATA-3851] Fix Data update issue with SCD/CDC for partition table
On spark
    cluster deploy mode
    
    Why is this PR needed?
    Issue 1:
    On process IUD with mergeDataSetCommand, we are setting direct write to store path carbon
property
    as true in executor side and resetting the value to false on driver side. Because of this,
Update
    operation is unsuccessfull, as Issue 2 will come into picture.
    
    Issue 2:
    Currently, For insert/load data into partition table, data file will be moved to partition
path
    and index files will remain in tmp file to improve commit job performance.
    When direct write to store path is enabled, data files are not moved to partition path
and load
    metrics are not set and added to task context. On commitJob, it just return, as index
and carbon
    size will be 0. Thus, the corresponding load is not successful.
    
    What changes were proposed in this PR?
    For Issue 1:
    Set direct write to store path carbon property to default value on close writer while
processIUD
    for merge data set command
    
    For Issue 2:
    1. While writing data file, check if partition table and dir path is "partition tmp path",
then
    write data file directly to partiiton path.
    2. On close writer, add partition carbon data and index info to metrics
    
    This closes #3835
---
 .../mutation/merge/CarbonMergeDataSetCommand.scala |  9 ++++-
 .../store/writer/AbstractFactDataWriter.java       | 46 ++++++++++++++++++++--
 2 files changed, 49 insertions(+), 6 deletions(-)

diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/mutation/merge/CarbonMergeDataSetCommand.scala
b/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/mutation/merge/CarbonMergeDataSetCommand.scala
index 4cee705..7abfc42 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/mutation/merge/CarbonMergeDataSetCommand.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/mutation/merge/CarbonMergeDataSetCommand.scala
@@ -180,8 +180,6 @@ case class CarbonMergeDataSetCommand(
     } else {
       None
     }
-    CarbonProperties.getInstance().addProperty(CarbonLoadOptionConstants
-      .ENABLE_CARBON_LOAD_DIRECT_WRITE_TO_STORE_PATH, "false")
 
     CarbonInsertIntoWithDf(
       databaseNameOp = Some(carbonTable.getDatabaseName),
@@ -271,6 +269,10 @@ case class CarbonMergeDataSetCommand(
     val config = SparkSQLUtil.broadCastHadoopConf(sparkSession.sparkContext, job.getConfiguration)
     (frame.rdd.coalesce(DistributionUtil.getConfiguredExecutors(sparkSession.sparkContext)).
       mapPartitionsWithIndex { case (index, iter) =>
+        var directlyWriteDataToHdfs = CarbonProperties.getInstance()
+          .getProperty(CarbonLoadOptionConstants
+            .ENABLE_CARBON_LOAD_DIRECT_WRITE_TO_STORE_PATH, CarbonLoadOptionConstants
+            .ENABLE_CARBON_LOAD_DIRECT_WRITE_TO_STORE_PATH_DEFAULT)
         CarbonProperties.getInstance().addProperty(CarbonLoadOptionConstants
           .ENABLE_CARBON_LOAD_DIRECT_WRITE_TO_STORE_PATH, "true")
         val confB = config.value.value
@@ -283,6 +285,9 @@ case class CarbonMergeDataSetCommand(
           val queue = new util.LinkedList[InternalRow]()
           override def hasNext: Boolean = if (!queue.isEmpty || iter.hasNext) true else {
             writer.close()
+            // revert load direct write to store path after insert
+            CarbonProperties.getInstance().addProperty(CarbonLoadOptionConstants
+              .ENABLE_CARBON_LOAD_DIRECT_WRITE_TO_STORE_PATH, directlyWriteDataToHdfs)
             false
           }
 
diff --git a/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
b/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
index 18911f4..637147c 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
@@ -281,12 +281,29 @@ public abstract class AbstractFactDataWriter implements CarbonFactDataWriter
{
         LOGGER.error(e);
       }
     } else {
-      if (currentFileSize == 0) {
-        try {
+      try {
+        if (currentFileSize == 0) {
           handleEmptyDataFile(carbonDataFileStorePath);
-        } catch (IOException e) {
-          LOGGER.error(e);
+        } else {
+          // In case if direct write data to store path is enabled, carbonData files will
be written
+          // directly to partition path. Need to add partition path and carbon data file
size info
+          // to load metrics
+          if (model.getTableSpec().getCarbonTable().isHivePartitionTable() && model
+              .getCarbonDataDirectoryPath().endsWith(".tmp") && carbonDataFileStorePath
+              .endsWith(CarbonCommonConstants.FACT_FILE_EXT)) {
+            if (metrics != null) {
+              // get partition directory path
+              String partitionPath = model.getCarbonDataDirectoryPath()
+                  .substring(0, model.getCarbonDataDirectoryPath().lastIndexOf(File.separator));
+              metrics.addToPartitionPath(partitionPath);
+              String carbonFilePath = partitionPath + carbonDataFileStorePath
+                  .substring(carbonDataFileStorePath.lastIndexOf(File.separator));
+              addOutputFilesInfoToMetrics(carbonFilePath);
+            }
+          }
         }
+      } catch (IOException e) {
+        LOGGER.error(e);
       }
     }
   }
@@ -318,6 +335,13 @@ public abstract class AbstractFactDataWriter implements CarbonFactDataWriter
{
     try {
       FileFactory.mkdirs(model.getCarbonDataDirectoryPath());
       if (enableDirectlyWriteDataToStorePath) {
+        if (model.getTableSpec().getCarbonTable().isHivePartitionTable() && model
+            .getCarbonDataDirectoryPath().endsWith(".tmp")) {
+          // set carbonData file store path to partition path instead of tmp directory
+          carbonDataFileStorePath = model.getCarbonDataDirectoryPath()
+              .substring(0, model.getCarbonDataDirectoryPath().lastIndexOf(File.separator))
+              + File.separator + carbonDataFileName;
+        }
         // the block size will be twice the block_size specified by user to make sure that
         // one carbondata file only consists exactly one HDFS block.
         fileOutputStream = FileFactory
@@ -440,9 +464,23 @@ public abstract class AbstractFactDataWriter implements CarbonFactDataWriter
{
           .copyCarbonDataFileToCarbonStorePath(indexFileName, model.getCarbonDataDirectoryPath(),
               fileSizeInBytes, metrics);
       FileFactory.deleteFile(indexFileName);
+    } else if (model.getTableSpec().getCarbonTable().isHivePartitionTable() && model
+        .getCarbonDataDirectoryPath().endsWith(".tmp")) {
+      if (metrics != null) {
+        addOutputFilesInfoToMetrics(indexFileName);
+      }
     }
   }
 
+  private void addOutputFilesInfoToMetrics(String carbonFilePath) {
+    // Store the number of files written by each task.
+    metrics.incrementCount();
+    // Store the files written by each task to metrics.
+    long targetSize = FileFactory.getCarbonFile(carbonFilePath).getSize();
+    metrics.addToOutputFiles(carbonFilePath + ":" + targetSize);
+    metrics.addOutputBytes(targetSize);
+  }
+
   /**
    * This method will close the executor service which is used for copying carbon
    * data files to carbon store path


Mime
View raw message