Merge pull request #21 from Ankitp1342/feature/autocorrect-rows

Auto-correct (enable/disable) missing or mismatched data

Author: Ankitp1342

pom.xml

@@ -3,7 +3,7 @@
 
   <groupId>com.datastax.spark.example</groupId>
   <artifactId>migrate</artifactId>
-  <version>0.13</version>
+  <version>0.15</version>
   <packaging>jar</packaging>
 
   <properties>

src/main/java/datastax/astra/migrate/AbstractJobSession.java

@@ -94,10 +94,16 @@ protected AbstractJobSession(CqlSession sourceSession, CqlSession astraSession,
             }
         }
 
-        minWriteTimeStampFilter = new Long(
-                sparkConf.get("spark.migrate.source.minWriteTimeStampFilter", "0"));
-        maxWriteTimeStampFilter = new Long(
-                sparkConf.get("spark.migrate.source.maxWriteTimeStampFilter", "" + Long.MAX_VALUE));
+        String minWriteTimeStampFilterStr =
+                sparkConf.get("spark.migrate.source.minWriteTimeStampFilter", "0");
+        if (null != minWriteTimeStampFilterStr && minWriteTimeStampFilterStr.trim().length() > 1) {
+            minWriteTimeStampFilter = Long.parseLong(minWriteTimeStampFilterStr);
+        }
+        String maxWriteTimeStampFilterStr =
+                sparkConf.get("spark.migrate.source.maxWriteTimeStampFilter", "0");
+        if (null != maxWriteTimeStampFilterStr && maxWriteTimeStampFilterStr.trim().length() > 1) {
+            maxWriteTimeStampFilter = Long.parseLong(maxWriteTimeStampFilterStr);
+        }
 
         logger.info(" DEFAULT -- Write Batch Size: " + batchSize);
         logger.info(" DEFAULT -- Source Keyspace Table: " + sourceKeyspaceTable);

src/main/java/datastax/astra/migrate/DiffJobSession.java

@@ -29,15 +29,16 @@ public class DiffJobSession extends CopyJobSession {
     private static DiffJobSession diffJobSession;
 
     private AtomicLong readCounter = new AtomicLong(0);
-    private AtomicLong diffCounter = new AtomicLong(0);
+    private AtomicLong mismatchCounter = new AtomicLong(0);
     private AtomicLong missingCounter = new AtomicLong(0);
     private AtomicLong correctedMissingCounter = new AtomicLong(0);
-    private AtomicLong validDiffCounter = new AtomicLong(0);
+    private AtomicLong correctedMismatchCounter = new AtomicLong(0);
+    private AtomicLong validCounter = new AtomicLong(0);
     private AtomicLong skippedCounter = new AtomicLong(0);
 
     protected List<MigrateDataType> selectColTypes = new ArrayList<MigrateDataType>();
-    protected Boolean isDiffOnly = false;
-
+    protected Boolean autoCorrectMissing = false;
+    protected Boolean autoCorrectMismatch = false;
 
     public static DiffJobSession getInstance(CqlSession sourceSession, CqlSession astraSession, SparkConf sparkConf) {
         if (diffJobSession == null) {
@@ -55,7 +56,8 @@ private DiffJobSession(CqlSession sourceSession, CqlSession astraSession, SparkC
         super(sourceSession, astraSession, sparkConf);
 
         selectColTypes = getTypes(sparkConf.get("spark.migrate.diff.select.types"));
-        isDiffOnly = Boolean.parseBoolean(sparkConf.get("spark.migrate.isDiffOnly", "false"));
+        autoCorrectMissing = Boolean.parseBoolean(sparkConf.get("spark.migrate.destination.autocorrect.missing", "false"));
+        autoCorrectMismatch = Boolean.parseBoolean(sparkConf.get("spark.migrate.destination.autocorrect.mismatch", "false"));
     }
 
     public void getDataAndDiff(BigInteger min, BigInteger max) {
@@ -105,14 +107,16 @@ public void getDataAndDiff(BigInteger min, BigInteger max) {
     public void printCounts(String finalStr) {
         logger.info("TreadID: " + Thread.currentThread().getId() + " " + finalStr + " Read Record Count: "
                 + readCounter.get());
-        logger.info("TreadID: " + Thread.currentThread().getId() + " " + finalStr + " Read Differences Count: "
-                + diffCounter.get());
-        logger.info("TreadID: " + Thread.currentThread().getId() + " " + finalStr + " Read Missing Count: "
+        logger.info("TreadID: " + Thread.currentThread().getId() + " " + finalStr + " Read Mismatch Count: "
+                + mismatchCounter.get());
+        logger.info("TreadID: " + Thread.currentThread().getId() + " " + finalStr + " Corrected Mismatch Count: "
+                + correctedMismatchCounter.get());
+                logger.info("TreadID: " + Thread.currentThread().getId() + " " + finalStr + " Read Missing Count: "
                 + missingCounter.get());
-        logger.info("TreadID: " + Thread.currentThread().getId() + " " + finalStr + " Read Corrected Missing Count: "
+        logger.info("TreadID: " + Thread.currentThread().getId() + " " + finalStr + " Corrected Missing Count: "
                 + correctedMissingCounter.get());
         logger.info("TreadID: " + Thread.currentThread().getId() + " " + finalStr + " Read Valid Count: "
-                + validDiffCounter.get());
+                + validCounter.get());
         logger.info("TreadID: " + Thread.currentThread().getId() + " " + finalStr + " Read Skipped Count: "
                 + skippedCounter.get());
     }
@@ -123,22 +127,30 @@ private void diff(Row sourceRow, Row astraRow) {
             logger.error("Data is missing in Astra: " + getKey(sourceRow));
             //correct data
 
-            if (!isDiffOnly) {
+            if (autoCorrectMissing) {
                 astraSession.execute(bindInsert(astraInsertStatement, sourceRow));
                 correctedMissingCounter.incrementAndGet();
                 logger.error("Corrected missing data in Astra: " + getKey(sourceRow));
             }
+
             return;
         }
 
         String diffData = isDifferent(sourceRow, astraRow);
         if (!diffData.isEmpty()) {
-            diffCounter.incrementAndGet();
-            logger.error("Data difference found -  Key: " + getKey(sourceRow) + " Data: " + diffData);
+            mismatchCounter.incrementAndGet();
+            logger.error("Data mismatch found -  Key: " + getKey(sourceRow) + " Data: " + diffData);
+
+            if (autoCorrectMismatch) {
+                astraSession.execute(bindInsert(astraInsertStatement, sourceRow));
+                correctedMismatchCounter.incrementAndGet();
+                logger.error("Corrected mismatch data in Astra: " + getKey(sourceRow));
+            }
+
             return;
         }
 
-        validDiffCounter.incrementAndGet();
+        validCounter.incrementAndGet();
     }
 
     private String isDifferent(Row sourceRow, Row astraRow) {

src/resources/sparkConf.properties

@@ -11,6 +11,8 @@ spark.migrate.destination.username                              client-id
 spark.migrate.destination.password                              client-secret
 spark.migrate.destination.read.consistency.level                LOCAL_QUORUM
 spark.migrate.destination.keyspaceTable                         test.a2
+spark.migrate.destination.autocorrect.missing                   false
+spark.migrate.destination.autocorrect.mismatch                  false
 
 spark.migrate.maxRetries                                        10
 spark.migrate.readRateLimit                                     40000