Merge pull request #92 from datastax/feature/rename-splitsize-to-numsplits

Renamed property splitSize to numSplits (support backward compatibility)

Author: pravinbhat

README.md

@@ -6,28 +6,21 @@ Migrate and Validate Tables between Origin and Target Cassandra Clusters.
 
 > :warning: Please note this job has been tested with spark version [3.3.1](https://archive.apache.org/dist/spark/spark-3.3.1/)
 
-## Container Image
+## Install as a Container
 - Get the latest image that includes all dependencies from [DockerHub](https://hub.docker.com/r/datastax/cassandra-data-migrator) 
-  - If you use this route, all migration tools (`cassandra-data-migrator` + `dsbulk` + `cqlsh`) would be available in the `/assets/` folder of the container
-- OR follow the below build steps (and Prerequisite) to build the jar locally
+  - All migration tools (`cassandra-data-migrator` + `dsbulk` + `cqlsh`) would be available in the `/assets/` folder of the container
 
-### Prerequisite
+## Install as a JAR file
+- Download the latest jar file from the GitHub [packages area here](https://github.com/orgs/datastax/packages?repo_name=cassandra-data-migrator)
 
+### Prerequisite
 - Install Java8 as spark binaries are compiled with it.
-- Install Maven 3.8.x
-- Install single instance of spark on a node where you want to run this job. Spark can be installed by running the following: -
-
+- Install Spark version [3.3.1](https://archive.apache.org/dist/spark/spark-3.3.1/) on a single VM (no cluster necessary) where you want to run this job. Spark can be installed by running the following: -
 ```
 wget https://archive.apache.org/dist/spark/spark-3.3.1/spark-3.3.1-bin-hadoop3.tgz
 tar -xvzf spark-3.3.1-bin-hadoop3.tgz
 ```
 
-### Build
-1. Clone this repo
-2. Move to the repo folder `cd cassandra-data-migrator`
-3. Run the build `mvn clean package`
-4. The fat jar (`cassandra-data-migrator-3.x.x.jar`) file should now be present in the `target` folder
-
 # Steps for Data-Migration:
 
 1. `sparkConf.properties` file needs to be configured as applicable for the environment
@@ -41,8 +34,14 @@ tar -xvzf spark-3.3.1-bin-hadoop3.tgz
 --class datastax.astra.migrate.Migrate cassandra-data-migrator-3.x.x.jar &> logfile_name.txt
 ```
 
-Note: Above command also generates a log file `logfile_name.txt` to avoid log output on the console.
-
+Note: 
+- Above command generates a log file `logfile_name.txt` to avoid log output on the console.
+- Add option `--driver-memory 25G --executor-memory 25G` as shown below if the table migrated is large (over 100GB)
+```
+./spark-submit --properties-file sparkConf.properties /
+--master "local[*]" --driver-memory 25G --executor-memory 25G /
+--class datastax.astra.migrate.Migrate cassandra-data-migrator-3.x.x.jar &> logfile_name.txt
+```
 
 # Steps for Data-Validation:
 
@@ -64,19 +63,20 @@ Note: Above command also generates a log file `logfile_name.txt` to avoid log ou
 ```
 
 - Please grep for all `ERROR` from the output log files to get the list of missing and mismatched records.
-  - Note that it lists differences by partition key values.
+  - Note that it lists differences by primary-key values.
 - The Validation job can also be run in an AutoCorrect mode. This mode can
   - Add any missing records from origin to target
-  - Fix any inconsistencies between origin and target (makes target same as origin). 
+  - Update any mismatched records between origin and target (makes target same as origin). 
 - Enable/disable this feature using one or both of the below setting in the config file
-
 ```
 spark.target.autocorrect.missing                    true|false
 spark.target.autocorrect.mismatch                   true|false
 ```
+Note:
+- The validation job will never delete records from target i.e. it only adds or updates data on target
 
 # Migrating specific partition ranges
-- You can also use the tool to migrate specific partition ranges, use class option `--class datastax.astra.migrate.MigratePartitionsFromFile` as shown below
+- You can also use the tool to migrate specific partition ranges using class option `--class datastax.astra.migrate.MigratePartitionsFromFile` as shown below
 ```
 ./spark-submit --properties-file sparkConf.properties /
 --master "local[*]" /
@@ -90,18 +90,26 @@ When running in above mode the tool assumes a `partitions.csv` file to be presen
 2637884402540451982,4638499294009575633
 798869613692279889,8699484505161403540
 ```
-This mode is specifically useful to processes a subset of partition-ranges that may have generated errors as a result of a previous long-running job to migrate a large table.
-
-# Additional features
-- [Counter tables](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_using/useCountersConcept.html)
-- Preserve [writetimes](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_reference/cql_commands/cqlSelect.html#cqlSelect__retrieving-the-datetime-a-write-occurred-p) and [TTL](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_reference/cql_commands/cqlSelect.html#cqlSelect__ref-select-ttl-p)
-- Advanced DataTypes ([Sets](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_reference/refDataTypes.html#refDataTypes__set), [Lists](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_reference/refDataTypes.html#refDataTypes__list), [Maps](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_reference/refDataTypes.html#refDataTypes__map), [UDTs](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_reference/refDataTypes.html#refDataTypes__udt))
-- Filter records from origin using writetimes, CQL conditions, token-ranges
+This mode is specifically useful to processes a subset of partition-ranges that may have failed during a previous run.
+
+# Features
+- Supports migration/validation of [Counter tables](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_using/useCountersConcept.html)
+- Preserve [writetimes](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_reference/cql_commands/cqlSelect.html#cqlSelect__retrieving-the-datetime-a-write-occurred-p) and [TTLs](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_reference/cql_commands/cqlSelect.html#cqlSelect__ref-select-ttl-p)
+- Supports migration/validation of advanced DataTypes ([Sets](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_reference/refDataTypes.html#refDataTypes__set), [Lists](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_reference/refDataTypes.html#refDataTypes__list), [Maps](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_reference/refDataTypes.html#refDataTypes__map), [UDTs](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_reference/refDataTypes.html#refDataTypes__udt))
+- Filter records from `Origin` using `writetimes` and/or CQL conditions and/or min/max token-range
+- Supports adding `constants` as new columns on `Target`
 - Fully containerized (Docker and K8s friendly)
 - SSL Support (including custom cipher algorithms)
-- Migrate from any Cassandra origin ([Apache Cassandra®](https://cassandra.apache.org) / [DataStax Enterprise™](https://www.datastax.com/products/datastax-enterprise) / [DataStax Astra DB™](https://www.datastax.com/products/datastax-astra)) to any Cassandra target ([Apache Cassandra®](https://cassandra.apache.org) / [DataStax Enterprise™](https://www.datastax.com/products/datastax-enterprise) / [DataStax Astra DB™](https://www.datastax.com/products/datastax-astra))
+- Migrate from any Cassandra `Origin` ([Apache Cassandra®](https://cassandra.apache.org) / [DataStax Enterprise™](https://www.datastax.com/products/datastax-enterprise) / [DataStax Astra DB™](https://www.datastax.com/products/datastax-astra)) to any Cassandra `Target` ([Apache Cassandra®](https://cassandra.apache.org) / [DataStax Enterprise™](https://www.datastax.com/products/datastax-enterprise) / [DataStax Astra DB™](https://www.datastax.com/products/datastax-astra))
+- Supports migration/validation from and to [Azure Cosmos Cassandra](https://learn.microsoft.com/en-us/azure/cosmos-db/cassandra)
 - Validate migration accuracy and performance using a smaller randomized data-set
-- Custom writetime
+- Supports adding custom fixed `writetime`
+
+# Building Jar for local development
+1. Clone this repo
+2. Move to the repo folder `cd cassandra-data-migrator`
+3. Run the build `mvn clean package` (Needs Maven 3.8.x)
+4. The fat jar (`cassandra-data-migrator-3.x.x.jar`) file should now be present in the `target` folder
 
 # Contributors
 Checkout all our wonderful contributors [here](./CONTRIBUTING.md#contributors).

pom.xml

@@ -8,7 +8,7 @@
 
   <properties>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-    <revision>3.2.1</revision>
+    <revision>3.2.2</revision>
     <scala.version>2.12.17</scala.version>
     <scala.main.version>2.12</scala.main.version>
     <spark.version>3.3.1</spark.version>

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

@@ -30,19 +30,19 @@ public static void main(String[] args) throws IOException {
         }
     }
 
-    public static Collection<Partition> getRandomSubPartitions(int splitSize, BigInteger min, BigInteger max, int coveragePercent) {
+    public static Collection<Partition> getRandomSubPartitions(int numSplits, BigInteger min, BigInteger max, int coveragePercent) {
         logger.info("ThreadID: {} Splitting min: {} max: {}", Thread.currentThread().getId(), min, max);
-        List<Partition> partitions = getSubPartitions(splitSize, min, max, coveragePercent);
+        List<Partition> partitions = getSubPartitions(numSplits, min, max, coveragePercent);
         Collections.shuffle(partitions);
         Collections.shuffle(partitions);
         Collections.shuffle(partitions);
         Collections.shuffle(partitions);
         return partitions;
     }
 
-    public static List<Partition> getSubPartitionsFromFile(int splitSize) throws IOException {
+    public static List<Partition> getSubPartitionsFromFile(int numSplits) throws IOException {
         logger.info("ThreadID: {} Splitting partitions in file: ./partitions.csv using a split-size of {}"
-                , Thread.currentThread().getId(), splitSize);
+                , Thread.currentThread().getId(), numSplits);
         List<Partition> partitions = new ArrayList<Partition>();
         BufferedReader reader = Util.getfileReader("./partitions.csv");
         String line = null;
@@ -52,7 +52,7 @@ public static List<Partition> getSubPartitionsFromFile(int splitSize) throws IOE
             }
             String[] minMax = line.split(",");
             try {
-                partitions.addAll(getSubPartitions(splitSize, new BigInteger(minMax[0]), new BigInteger(minMax[1]), 100));
+                partitions.addAll(getSubPartitions(numSplits, new BigInteger(minMax[0]), new BigInteger(minMax[1]), 100));
             } catch (Exception e) {
                 logger.error("Skipping partition: {}", line, e);
             }
@@ -61,9 +61,9 @@ public static List<Partition> getSubPartitionsFromFile(int splitSize) throws IOE
         return partitions;
     }
 
-    public static List<PKRows> getRowPartsFromFile(int splitSize) throws IOException {
+    public static List<PKRows> getRowPartsFromFile(int numSplits) throws IOException {
         logger.info("ThreadID: {} Splitting rows in file: ./primary_key_rows.csv using a split-size of {}"
-                , Thread.currentThread().getId(), splitSize);
+                , Thread.currentThread().getId(), numSplits);
         List<String> pkRows = new ArrayList<String>();
         BufferedReader reader = Util.getfileReader("./primary_key_rows.csv");
         String pkRow = null;
@@ -73,7 +73,7 @@ public static List<PKRows> getRowPartsFromFile(int splitSize) throws IOException
             }
             pkRows.add(pkRow);
         }
-        int partSize = pkRows.size() / splitSize;
+        int partSize = pkRows.size() / numSplits;
         if (partSize == 0) {
             partSize = pkRows.size();
         }
@@ -91,12 +91,12 @@ public static <T> Stream<List<T>> batches(List<T> source, int length) {
                 n -> source.subList(n * length, n == fullChunks ? size : (n + 1) * length));
     }
 
-    private static List<Partition> getSubPartitions(int splitSize, BigInteger min, BigInteger max, int coveragePercent) {
+    private static List<Partition> getSubPartitions(int numSplits, BigInteger min, BigInteger max, int coveragePercent) {
         if (coveragePercent < 1 || coveragePercent > 100) {
             coveragePercent = 100;
         }
         BigInteger curMax = new BigInteger(min.toString());
-        BigInteger partitionSize = max.subtract(min).divide(BigInteger.valueOf(splitSize));
+        BigInteger partitionSize = max.subtract(min).divide(BigInteger.valueOf(numSplits));
         List<Partition> partitions = new ArrayList<Partition>();
         if (partitionSize.compareTo(new BigInteger("0")) == 0) {
             partitionSize = new BigInteger("100000");

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

@@ -7,7 +7,7 @@ class AbstractJob extends BaseJob {
 
   abstractLogger.info("PARAM -- Min Partition: " + minPartition)
   abstractLogger.info("PARAM -- Max Partition: " + maxPartition)
-  abstractLogger.info("PARAM -- Split Size: " + splitSize)
+  abstractLogger.info("PARAM -- Number of Splits : " + numSplits)
   abstractLogger.info("PARAM -- Coverage Percent: " + coveragePercent)
   abstractLogger.info("PARAM -- Origin SSL Enabled: {}", sourceSSLEnabled);
   abstractLogger.info("PARAM -- Target SSL Enabled: {}", destinationSSLEnabled);

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

@@ -49,7 +49,8 @@ class BaseJob extends App {
   val minPartition = new BigInteger(Util.getSparkPropOr(sc, "spark.origin.minPartition", "-9223372036854775808"))
   val maxPartition = new BigInteger(Util.getSparkPropOr(sc, "spark.origin.maxPartition", "9223372036854775807"))
   val coveragePercent = Util.getSparkPropOr(sc, "spark.coveragePercent", "100")
-  val splitSize = Integer.parseInt(Util.getSparkPropOr(sc, "spark.splitSize", "10000"))
+  val splitSizeBackwardCompatibility = Util.getSparkPropOr(sc, "spark.splitSize", "10000")
+  val numSplits = Integer.parseInt(Util.getSparkPropOr(sc, "spark.numSplits", splitSizeBackwardCompatibility))
 
   protected def exitSpark() = {
     spark.stop()

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

@@ -16,7 +16,7 @@ object DiffData extends AbstractJob {
   exitSpark
 
   private def diffTable(sourceConnection: CassandraConnector, destinationConnection: CassandraConnector, config: SparkConf) = {
-    val partitions = SplitPartitions.getRandomSubPartitions(splitSize, minPartition, maxPartition, Integer.parseInt(coveragePercent))
+    val partitions = SplitPartitions.getRandomSubPartitions(numSplits, minPartition, maxPartition, Integer.parseInt(coveragePercent))
     logger.info("PARAM Calculated -- Total Partitions: " + partitions.size())
     val parts = sContext.parallelize(partitions.toSeq, partitions.size);
     logger.info("Spark parallelize created : " + parts.count() + " parts!");

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

@@ -18,7 +18,7 @@ object Migrate extends AbstractJob {
   exitSpark
 
   private def migrateTable(sourceConnection: CassandraConnector, destinationConnection: CassandraConnector, config: SparkConf) = {
-    val partitions = SplitPartitions.getRandomSubPartitions(splitSize, minPartition, maxPartition, Integer.parseInt(coveragePercent))
+    val partitions = SplitPartitions.getRandomSubPartitions(numSplits, minPartition, maxPartition, Integer.parseInt(coveragePercent))
     logger.info("PARAM Calculated -- Total Partitions: " + partitions.size())
     val parts = sContext.parallelize(partitions.toSeq, partitions.size);
     logger.info("Spark parallelize created : " + parts.count() + " parts!");

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

@@ -15,7 +15,7 @@ object MigratePartitionsFromFile extends AbstractJob {
   exitSpark
 
   private def migrateTable(sourceConnection: CassandraConnector, destinationConnection: CassandraConnector) = {
-    val partitions = SplitPartitions.getSubPartitionsFromFile(splitSize)
+    val partitions = SplitPartitions.getSubPartitionsFromFile(numSplits)
     logger.info("PARAM Calculated -- Total Partitions: " + partitions.size())
     val parts = sContext.parallelize(partitions.toSeq, partitions.size);
     logger.info("Spark parallelize created : " + parts.count() + " parts!");

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

@@ -13,7 +13,7 @@ object MigrateRowsFromFile extends AbstractJob {
   exitSpark
 
   private def migrateTable(sourceConnection: CassandraConnector, destinationConnection: CassandraConnector) = {
-    val listOfPKRows = SplitPartitions.getRowPartsFromFile(splitSize)
+    val listOfPKRows = SplitPartitions.getRowPartsFromFile(numSplits)
     logger.info("PARAM Calculated -- Number of PKRows: " + listOfPKRows.size())
 
     sourceConnection.withSessionDo(sourceSession =>

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

@@ -63,7 +63,7 @@ object OriginData extends BaseJob {
   }
 
   private def analyzeSourceTable(sourceConnection: CassandraConnector) = {
-    val partitions = SplitPartitions.getRandomSubPartitions(splitSize, minPartition, maxPartition, Integer.parseInt(coveragePercent))
+    val partitions = SplitPartitions.getRandomSubPartitions(numSplits, minPartition, maxPartition, Integer.parseInt(coveragePercent))
     logger.info("PARAM Calculated -- Total Partitions: " + partitions.size())
     val parts = sContext.parallelize(partitions.toSeq, partitions.size);
     logger.info("Spark parallelize created : " + parts.count() + " parts!");