Moved filtering properties to not be required in original config

Author: mfmaher2

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

@@ -23,10 +23,10 @@ public class CopyJobSession extends AbstractJobSession {
     protected CopyJobSession(CqlSession sourceSession, CqlSession astraSession, SparkConf sc) {
         super(sourceSession, astraSession, sc);
         filterData = Boolean.parseBoolean(sc.get("spark.origin.FilterData", "false"));
-        filterColName = sc.get("spark.origin.FilterColumn");
-        filterColType = sc.get("spark.origin.FilterColumnType");
+        filterColName = Util.getSparkPropOrEmpty(sc, "spark.origin.FilterColumn");
+        filterColType = Util.getSparkPropOrEmpty(sc, "spark.origin.FilterColumnType");
         filterColIndex = Integer.parseInt(sc.get("spark.origin.FilterColumnIndex", "0"));
-        filterColValue = sc.get("spark.origin.FilterColumnValue");
+        filterColValue = Util.getSparkPropOrEmpty(sc, "spark.origin.FilterColumnValue");
     }
 
     public static CopyJobSession getInstance(CqlSession sourceSession, CqlSession astraSession, SparkConf sc) {
@@ -60,7 +60,8 @@ public void getDataAndInsert(BigInteger min, BigInteger max) {
                         if (filterData) {
                             String col = (String) getData(new MigrateDataType(filterColType), filterColIndex, sourceRow);
                             if (col.trim().equalsIgnoreCase(filterColValue)) {
-                                logger.warn("Row larger than 10 MB found filtering out: " + getKey(sourceRow));
+                                logger.warn("Skipping row and filtering out: " + getKey(sourceRow));
+                                skippedCounter.incrementAndGet();
                                 continue;
                             }
                         }
@@ -108,7 +109,8 @@ public void getDataAndInsert(BigInteger min, BigInteger max) {
                         if (filterData) {
                             String colValue = (String) getData(new MigrateDataType(filterColType), filterColIndex, sourceRow);
                             if (colValue.trim().equalsIgnoreCase(filterColValue)) {
-                                logger.warn("Row larger than 10 MB found filtering out: " + getKey(sourceRow));
+                                logger.warn("Skipping row and filtering out: " + getKey(sourceRow));
+                                skippedCounter.incrementAndGet();
                                 continue;
                             }
                         }

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

@@ -43,8 +43,8 @@ protected OriginCountJobSession(CqlSession sourceSession, SparkConf sparkConf) {
         checkTableforColSize = Boolean.parseBoolean(sparkConf.get("spark.origin.checkTableforColSize", "false"));
         checkTableforselectCols = sparkConf.get("spark.origin.checkTableforColSize.cols");
         checkTableforColSizeTypes = getTypes(sparkConf.get("spark.origin.checkTableforColSize.cols.types"));
-        filterColName = sparkConf.get("spark.origin.FilterColumn");
-        filterColType = sparkConf.get("spark.origin.FilterColumnType");
+        filterColName = Util.getSparkPropOrEmpty(sparkConf, "spark.origin.FilterColumn");
+        filterColType = Util.getSparkPropOrEmpty(sparkConf, "spark.origin.FilterColumnType");
         filterColIndex = Integer.parseInt(sparkConf.get("spark.origin.FilterColumnIndex", "0"));
         fieldGuardraillimitMB = Integer.parseInt(sparkConf.get("spark.fieldGuardraillimitMB", "0"));
 

src/main/scala/datastax/astra/migrate/BaseJob.scala

@@ -49,6 +49,15 @@ class BaseJob extends App {
   val coveragePercent = Util.getSparkPropOr(sc, "spark.coveragePercent", "100")
   val splitSize = Integer.parseInt(Util.getSparkPropOr(sc, "spark.splitSize", "10000"))
 
+  val checkTableforColSize = Util.getSparkPropOr(sc, "spark.origin.checkTableforColSize", "false")
+  val checkTableforselectCols = Util.getSparkPropOrEmpty(sc, "spark.origin.checkTableforColSize.cols")
+//  val checkTableforColSizeTypes = getTypes(Util.getSparkPropOr(sc, "spark.origin.checkTableforColSize.cols.types"))
+  val filterColName = Util.getSparkPropOrEmpty(sc, "spark.origin.FilterColumn")
+  val filterColType = Util.getSparkPropOrEmpty(sc, "spark.origin.FilterColumnType")
+  val filterColIndex = Util.getSparkPropOr(sc, "spark.origin.FilterColumnIndex", "0")
+  val fieldGuardraillimitMB = Util.getSparkPropOr(sc, "spark.fieldGuardraillimitMB", "0")
+
+
   protected def exitSpark() = {
     spark.stop()
     abstractLogger.info("################################################################################################")

src/resources/sparkConf.properties

@@ -4,14 +4,6 @@ spark.origin.username                             some-username
 spark.origin.password                             some-secret-password
 spark.origin.read.consistency.level               LOCAL_QUORUM
 spark.origin.keyspaceTable                        test.a1
-spark.origin.checkTableforColSize                 false
-spark.origin.checkTableforColSize.cols            partition-key,clustering-key
-spark.origin.checkTableforColSize.cols.types      9,1
-spark.origin.FilterData                           true
-spark.origin.FilterColumn                         test
-spark.origin.FilterColumnIndex                    2
-spark.origin.FilterColumnType                     6%16
-spark.origin.FilterColumnValue                    test
 
 spark.target.isAstra                              true
 spark.target.scb                                  file:///aaa/bbb/secure-connect-enterprise.zip
@@ -48,6 +40,18 @@ spark.origin.writeTimeStampFilter                 false
 spark.origin.minWriteTimeStampFilter              0
 spark.origin.maxWriteTimeStampFilter              9223372036854775807
 
+################### ONLY USE if needing to get record count of recs greater than 10MB from Origin ######################
+#spark.origin.checkTableforColSize                 false
+#spark.origin.checkTableforColSize.cols            partition-key,clustering-key
+#spark.origin.checkTableforColSize.cols.types      9,1
+
+########################## ONLY USE if needing to filter data from Origin ###############################
+#spark.origin.FilterData                           false
+#spark.origin.FilterColumn                         test
+#spark.origin.FilterColumnIndex                    2
+#spark.origin.FilterColumnType                     6%16
+#spark.origin.FilterColumnValue                    test
+
 ########################## ONLY USE if SSL clientAuth is enabled on origin Cassandra/DSE ###############################
 #spark.origin.trustStore.path
 #spark.origin.trustStore.password