Exception handler - Partition level

Author: faizalrub-datastax

SIT/smoke/01_basic_kvp/cdm.sh

@@ -10,7 +10,7 @@ EOF
 spark-submit \
   --properties-file /smoke/01_basic_kvp/migrate.properties \
   --master "local[*]" \
-  --class datastax.astra.migrate.Migrate /local/cassandra-data-migrator.jar 
+  --class datastax.cdm.job.Migrate /local/cassandra-data-migrator.jar
 
 cat <<EOF
 !!!!!!!!
@@ -21,4 +21,4 @@ EOF
 spark-submit \
   --properties-file /smoke/01_basic_kvp/migrate.properties \
   --master "local[*]" \
-  --class datastax.astra.migrate.DiffData /local/cassandra-data-migrator.jar
+  --class datastax.cdm.job.DiffData /local/cassandra-data-migrator.jar

src/main/java/datastax/cdm/job/AbstractJobSession.java

@@ -43,6 +43,8 @@ protected AbstractJobSession(CqlSession originSession, CqlSession targetSession,
         logger.info("PARAM -- Max Retries: {}", maxRetries);
         logger.info("PARAM -- ReadRateLimit: {}", readLimiter.getRate());
         logger.info("PARAM -- WriteRateLimit: {}", writeLimiter.getRate());
+        logger.info("PARAM -- Token range exception dir: {}", tokenRangeExceptionDir);
+        logger.info("PARAM -- Token range exception file name: {}", exceptionFileName);
 
         cqlHelper.initialize();
     }

src/main/java/datastax/cdm/job/DiffJobSession.java

@@ -205,6 +205,7 @@ private void diff(Record record) {
             //correct data
             if (autoCorrectMissing) {
                 writeLimiter.acquire(1);
+                logger.info("Inserting missing row in target: {}", record.getPk());
                 if (isCounterTable) cqlHelper.getTargetUpdateStatement().putRecord(record);
                 else cqlHelper.getTargetInsertStatement().putRecord(record);
                 correctedMissingCounter.incrementAndGet();
@@ -220,6 +221,7 @@ private void diff(Record record) {
 
             if (autoCorrectMismatch) {
                 writeLimiter.acquire(1);
+                logger.info("Correcting mismatch row in target: {}", record.getPk());
                 if (isCounterTable) cqlHelper.getTargetUpdateStatement().putRecord(record);
                 else cqlHelper.getTargetInsertStatement().putRecord(record);
                 correctedMismatchCounter.incrementAndGet();

src/main/java/datastax/cdm/properties/KnownProperties.java

@@ -172,8 +172,8 @@ public enum PropertyType {
     //==========================================================================
     // Error handling
     //==========================================================================
-    public static final String TOKEN_RANGE_EXCEPTION_DIR = "spark.tokenRange.exceptionDir"; // file:///aaa/bbb/
-    public static final String TOKEN_RANGE_EXCEPTION_FILE = "spark.input.partitionFile"; // file:///aaa/bbb/partitions.csv
+    public static final String TOKEN_RANGE_EXCEPTION_DIR = "spark.tokenRange.exceptionDir"; // aaa/bbb
+    public static final String TOKEN_RANGE_EXCEPTION_FILE = "spark.input.partitionFile"; // aaa/bbb/filename
     static {
             types.put(TOKEN_RANGE_EXCEPTION_DIR, PropertyType.STRING);
             types.put(TOKEN_RANGE_EXCEPTION_FILE, PropertyType.STRING);

src/main/scala/datastax/cdm/job/DiffDataFailedPartitionsFromFile.scala

@@ -1,6 +1,7 @@
 package datastax.cdm.job
 
 import com.datastax.spark.connector.cql.CassandraConnector
+import org.apache.spark.SparkConf
 import org.slf4j.LoggerFactory
 
 import scala.collection.JavaConversions._
@@ -10,11 +11,11 @@ object DiffDataFailedPartitionsFromFile extends AbstractJob {
   val logger = LoggerFactory.getLogger(this.getClass.getName)
   logger.info("Started MigratePartitionsFromFile App")
 
-  migrateTable(originConnection, targetConnection)
+  migrateTable(originConnection, targetConnection, sc)
 
   exitSpark
 
-  private def migrateTable(sourceConnection: CassandraConnector, destinationConnection: CassandraConnector) = {
+  private def migrateTable(sourceConnection: CassandraConnector, destinationConnection: CassandraConnector, config: SparkConf) = {
     val partitions = SplitPartitions.getFailedSubPartitionsFromFile(numSplits, tokenRangeFile)
     logger.info("PARAM Calculated -- Total Partitions: " + partitions.size())
     val parts = sContext.parallelize(partitions.toSeq, partitions.size);
@@ -23,11 +24,11 @@ object DiffDataFailedPartitionsFromFile extends AbstractJob {
     parts.foreach(part => {
       sourceConnection.withSessionDo(sourceSession =>
         destinationConnection.withSessionDo(destinationSession =>
-          DiffJobSession.getInstance(sourceSession, destinationSession, sc)
+          DiffJobSession.getInstance(sourceSession, destinationSession, config)
             .getDataAndDiff(part.getMin, part.getMax)))
     })
 
-    DiffJobSession.getInstance(null, null, sc).printCounts(true);
+    DiffJobSession.getInstance(null, null, config).printCounts(true);
   }
 
 }

src/resources/runCommands.txt

@@ -4,11 +4,14 @@ curl -OL https://downloads.datastax.com/enterprise/cqlsh-astra.tar.gz
 wget https://archive.apache.org/dist/spark/spark-3.3.1/spark-3.3.1-bin-hadoop3.tgz
 
 // Migrate
-spark-submit --properties-file /<path>/sparkConf.properties --verbose --master "local[*]" --class datastax.astra.migrate.Migrate /<path>/cassandra-data-migrator-3.*.jar
-spark-submit --properties-file /<path>/sparkConf.properties --master "local[*]" --driver-memory 25G --executor-memory 25G --class datastax.astra.migrate.Migrate /<path>/cassandra-data-migrator-3.*.jar &> table_out.log
+spark-submit --properties-file /<path>/sparkConf.properties --verbose --master "local[*]" --class datastax.cdm.job.Migrate /<path>/cassandra-data-migrator-3.*.jar
+spark-submit --properties-file /<path>/sparkConf.properties --master "local[*]" --driver-memory 25G --executor-memory 25G --class datastax.cdm.job.Migrate /<path>/cassandra-data-migrator-3.*.jar &> table_out.log
 
 // Random Partitioner Run Command
-spark-submit --properties-file /<path>/sparkConf.properties --verbose --master "local[*]" --conf spark.origin.minPartition=-1 --conf spark.origin.maxPartition=170141183460469231731687303715884105728 --class datastax.astra.migrate.Migrate /<path>/cassandra-data-migrator-3.*.jar
+spark-submit --properties-file /<path>/sparkConf.properties --verbose --master "local[*]" --conf spark.origin.minPartition=-1 --conf spark.origin.maxPartition=170141183460469231731687303715884105728 --class datastax.cdm.job.Migrate /<path>/cassandra-data-migrator-3.*.jar
 
 // Validate
-spark-submit --properties-file /<path>/sparkConf.properties --master "local[*]" --driver-memory 25G --executor-memory 25G --class datastax.astra.migrate.DiffData /<path>/cassandra-data-migrator-3.*.jar &> table_out.log
+spark-submit --properties-file /<path>/sparkConf.properties --master "local[*]" --driver-memory 25G --executor-memory 25G --class datastax.cdm.job.DiffData /<path>/cassandra-data-migrator-3.*.jar &> table_out.log
+
+// Rerun validate with failed partition ranges
+spark-submit --properties-file /<path>/sparkConf.properties --master "local[*]" --conf spark.input.partitionFile=<path>>/<fileName>  --driver-memory 25G --executor-memory 25G --class datastax.cdm.job.DiffDataFailedPartitionsFromFile /<path>/cassandra-data-migrator-3.*.jar &> table_out.log

src/resources/sparkConf.properties

@@ -164,12 +164,15 @@ spark.cdm.schema.target.column.id.names           partition_col1,partition_col2,
 #                           5323. The corresponding counter in Origin is also 5323. At some point, the Target
 #                           counter gets DELETEd. Should the .missing record be re-inserted before
 #                           the DELETE gets tombstoned, the counter will zombie back to life, and the
-#                           counter will become 5323+5323 = 10646. 
+#                           counter will become 5323+5323 = 10646.
+# spark.tokenRange
+#   .exceptionDir         : When set partition ranges corresponding to the exceptions will be logged in
+#                           file with the name <keyspace>.<table_name> in this directory.
 #-----------------------------------------------------------------------------------------------------------
 spark.cdm.autocorrect.missing                     false
 spark.cdm.autocorrect.mismatch                    false
 #spark.cdm.autocorrect.missing.counter             false
-
+#spark.tokenRange.exceptionDir                      aaa/bbb
 #===========================================================================================================
 # Performance and Operations Parameters affecting throughput, error handling, and similar concerns.
 #