Merge pull request #120 from datastax/issue/CDM-30

Issue/cdm 30

Author: mieslep

README.md

@@ -24,15 +24,17 @@ tar -xvzf spark-3.3.1-bin-hadoop3.tgz
 
 # Steps for Data-Migration:
 
-1. `sparkConf.properties` file needs to be configured as applicable for the environment
+> :warning: Note that Version 4 of the tool is not backward-compatible with .properties files created in previous versions, and that package names have changed.
+
+1. `sparkConf.properties` file needs to be configured as applicable for the environment. Parameter descriptions and defaults are described in the file.
    > A sample Spark conf file configuration can be [found here](./src/resources/sparkConf.properties)
 2. Place the conf file where it can be accessed while running the job via spark-submit.
 3. Run the below job using `spark-submit` command as shown below:
 
 ```
 ./spark-submit --properties-file sparkConf.properties /
 --master "local[*]" /
---class datastax.astra.migrate.Migrate cassandra-data-migrator-3.x.x.jar &> logfile_name.txt
+--class datastax.cdm.job.Migrate cassandra-data-migrator-4.x.x.jar &> logfile_name.txt
 ```
 
 Note: 
@@ -41,26 +43,26 @@ Note:
 ```
 ./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
+--class datastax.cdm.job.Migrate cassandra-data-migrator-4.x.x.jar &> logfile_name.txt
 ```
 
 # Steps for Data-Validation:
 
-- To run the job in Data validation mode, use class option `--class datastax.astra.migrate.DiffData` as shown below
+- To run the job in Data validation mode, use class option `--class datastax.cdm.job.DiffData` as shown below
 
 ```
 ./spark-submit --properties-file sparkConf.properties /
 --master "local[*]" /
---class datastax.astra.migrate.DiffData cassandra-data-migrator-3.x.x.jar &> logfile_name.txt
+--class datastax.cdm.job.DiffData cassandra-data-migrator-4.x.x.jar &> logfile_name.txt
 ```
 
 - Validation job will report differences as “ERRORS” in the log file as shown below
 
 ```
-22/10/27 23:25:29 ERROR DiffJobSession: Missing target row found for key: Grapes %% 1 %% 2020-05-22 %% 2020-05-23T00:05:09.353Z %% skuid %% Aliquam faucibus
-22/10/27 23:25:29 ERROR DiffJobSession: Inserted missing row in target: Grapes %% 1 %% 2020-05-22 %% 2020-05-23T00:05:09.353Z %% skuid %% Aliquam faucibus
-22/10/27 23:25:30 ERROR DiffJobSession: Mismatch row found for key: Grapes %% 1 %% 2020-05-22 %% 2020-05-23T00:05:09.353Z %% skuid %% augue odio at quam Data:  (Index: 8 Origin: Hello 3 Target: Hello 2 )
-22/10/27 23:25:30 ERROR DiffJobSession: Updated mismatch row in target: Grapes %% 1 %% 2020-05-22 %% 2020-05-23T00:05:09.353Z %% skuid %% augue odio at quam
+23/04/06 08:43:06 ERROR DiffJobSession: Mismatch row found for key: [key3] Mismatch: Target Index: 1 Origin: valueC Target: value999) 
+23/04/06 08:43:06 ERROR DiffJobSession: Corrected mismatch row in target: [key3]
+23/04/06 08:43:06 ERROR DiffJobSession: Missing target row found for key: [key2]
+23/04/06 08:43:06 ERROR DiffJobSession: Inserted missing row in target: [key2]
 ```
 
 - Please grep for all `ERROR` from the output log files to get the list of missing and mismatched records.
@@ -70,18 +72,18 @@ Note:
   - 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
+spark.cdm.autocorrect.missing                     false|true
+spark.cdm.autocorrect.mismatch                    false|true
 ```
 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 using 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.cdm.job.MigratePartitionsFromFile` as shown below
 ```
 ./spark-submit --properties-file sparkConf.properties /
 --master "local[*]" /
---class datastax.astra.migrate.MigratePartitionsFromFile cassandra-data-migrator-3.x.x.jar &> logfile_name.txt
+--class datastax.cdm.job.MigratePartitionsFromFile cassandra-data-migrator-4.x.x.jar &> logfile_name.txt
 ```
 
 When running in above mode the tool assumes a `partitions.csv` file to be present in the current folder in the below format, where each line (`min,max`) represents a partition-range 
@@ -99,6 +101,7 @@ This mode is specifically useful to processes a subset of partition-ranges that
 - 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`
+- Supports expanding `Map` columns on `Origin` into multiple records 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))
@@ -110,7 +113,7 @@ This mode is specifically useful to processes a subset of partition-ranges that
 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
+4. The fat jar (`cassandra-data-migrator-4.x.x.jar`) file should now be present in the `target` folder
 
 # Contributors
 Checkout all our wonderful contributors [here](./CONTRIBUTING.md#contributors).

SIT/common.sh

@@ -2,7 +2,7 @@ export DOCKER_CASS=cdm-sit-cass
 export DOCKER_CDM=cdm-sit-cdm
 export CASS_USERNAME=cassandra
 export CASS_PASSWORD=cassandra
-export KEYSPACES="source target"
+export KEYSPACES="origin target"
 export CDM_DIRECTORY=/local
 export CDM_JARFILE=cassandra-data-migrator.jar
 

SIT/features/01_constant_column/cdm.txt

@@ -1,3 +1,3 @@
-migrateData datastax.astra.migrate.Migrate migrate.properties
-validateData datastax.astra.migrate.DiffData migrate.properties
-fixData datastax.astra.migrate.DiffData fix.properties
+migrateData datastax.cdm.job.Migrate migrate.properties
+validateData datastax.cdm.job.DiffData migrate.properties
+fixData datastax.cdm.job.DiffData fix.properties

SIT/features/01_constant_column/fix.properties

@@ -1,25 +1,18 @@
-spark.origin.host                                 cdm-sit-cass
-spark.origin.port                                 9042
-spark.origin.username                             cassandra
-spark.origin.password                             cassandra
-spark.origin.keyspaceTable                        source.feature_constant_column
+spark.cdm.origin.connect.host                     cdm-sit-cass
+spark.cdm.target.connect.host                     cdm-sit-cass
 
-spark.target.host                                 cdm-sit-cass
-spark.target.port                                 9042
-spark.target.username                             cassandra
-spark.target.password                             cassandra
-spark.target.keyspaceTable                        target.feature_constant_column
+spark.cdm.schema.origin.keyspaceTable             origin.feature_constant_column
+spark.cdm.schema.target.keyspaceTable             target.feature_constant_column
+spark.cdm.perfops.numParts                        1
 
-spark.numSplits                                   1
+spark.cdm.schema.origin.column.names              key,value
+spark.cdm.schema.origin.column.partition.names    key
+spark.cdm.schema.target.column.id.names           const1,key
+spark.cdm.schema.origin.column.types              0,0
 
-spark.query.origin                                key,value
-spark.query.origin.partitionKey                   key
-spark.query.target.id                             const1,key
-spark.query.types                                 0,0
+spark.cdm.feature.constantColumns.names           const1,const2
+spark.cdm.feature.constantColumns.types           0,1
+spark.cdm.feature.constantColumns.values          'abcd',1234
 
-spark.cdm.cql.feature.constantColumns.names       const1,const2
-spark.cdm.cql.feature.constantColumns.types       0,1
-spark.cdm.cql.feature.constantColumns.values      'abcd',1234
-
-spark.target.autocorrect.missing                  true
-spark.target.autocorrect.mismatch                 true
+spark.cdm.autocorrect.missing                     true
+spark.cdm.autocorrect.mismatch                    true

SIT/features/01_constant_column/migrate.properties

@@ -1,22 +1,16 @@
-spark.origin.host                                 cdm-sit-cass
-spark.origin.port                                 9042
-spark.origin.username                             cassandra
-spark.origin.password                             cassandra
-spark.origin.keyspaceTable                        source.feature_constant_column
+spark.cdm.origin.connect.host                     cdm-sit-cass
+spark.cdm.target.connect.host                     cdm-sit-cass
 
-spark.target.host                                 cdm-sit-cass
-spark.target.port                                 9042
-spark.target.username                             cassandra
-spark.target.password                             cassandra
-spark.target.keyspaceTable                        target.feature_constant_column
+spark.cdm.schema.origin.keyspaceTable             origin.feature_constant_column
+spark.cdm.schema.target.keyspaceTable             target.feature_constant_column
+spark.cdm.perfops.numParts                        1
 
-spark.numSplits                                   1
+spark.cdm.schema.origin.column.names              key,value
+spark.cdm.schema.origin.column.partition.names    key
+spark.cdm.schema.target.column.id.names           const1,key
+spark.cdm.schema.origin.column.types              0,0
 
-spark.query.origin                                key,value
-spark.query.origin.partitionKey                   key
-spark.query.target.id                             const1,key
-spark.query.types                                 0,0
+spark.cdm.feature.constantColumns.names           const1,const2
+spark.cdm.feature.constantColumns.types           0,1
+spark.cdm.feature.constantColumns.values          'abcd',1234
 
-spark.cdm.cql.feature.constantColumns.names       const1,const2
-spark.cdm.cql.feature.constantColumns.types       0,1
-spark.cdm.cql.feature.constantColumns.values      'abcd',1234

SIT/features/01_constant_column/setup.cql

@@ -1,8 +1,8 @@
-DROP TABLE IF EXISTS source.feature_constant_column;
-CREATE TABLE source.feature_constant_column(key text, value text, PRIMARY KEY (key));
-INSERT INTO source.feature_constant_column(key,value) VALUES ('key1','valueA');
-INSERT INTO source.feature_constant_column(key,value) VALUES ('key2','valueB');
-INSERT INTO source.feature_constant_column(key,value) VALUES ('key3','valueC');
+DROP TABLE IF EXISTS origin.feature_constant_column;
+CREATE TABLE origin.feature_constant_column(key text, value text, PRIMARY KEY (key));
+INSERT INTO origin.feature_constant_column(key,value) VALUES ('key1','valueA');
+INSERT INTO origin.feature_constant_column(key,value) VALUES ('key2','valueB');
+INSERT INTO origin.feature_constant_column(key,value) VALUES ('key3','valueC');
 
 DROP TABLE IF EXISTS target.feature_constant_column;
 CREATE TABLE target.feature_constant_column(const1 text, key text, value text, const2 int, PRIMARY KEY (const1, key));

SIT/features/02_explode_map/cdm.txt

@@ -1,3 +1,3 @@
-migrateData datastax.astra.migrate.Migrate migrate.properties
-validateData datastax.astra.migrate.DiffData migrate.properties
-fixData datastax.astra.migrate.DiffData fix.properties
+migrateData datastax.cdm.job.Migrate migrate.properties
+validateData datastax.cdm.job.DiffData migrate.properties
+fixData datastax.cdm.job.DiffData fix.properties

SIT/features/02_explode_map/fix.properties

@@ -1,25 +1,19 @@
-spark.origin.host                                 cdm-sit-cass
-spark.origin.port                                 9042
-spark.origin.username                             cassandra
-spark.origin.password                             cassandra
-spark.origin.keyspaceTable                        source.feature_explode_map
+spark.cdm.origin.connect.host                     cdm-sit-cass
+spark.cdm.target.connect.host                     cdm-sit-cass
 
-spark.target.host                                 cdm-sit-cass
-spark.target.port                                 9042
-spark.target.username                             cassandra
-spark.target.password                             cassandra
-spark.target.keyspaceTable                        target.feature_explode_map
+spark.cdm.schema.origin.keyspaceTable             origin.feature_explode_map
+spark.cdm.schema.target.keyspaceTable             target.feature_explode_map
+spark.cdm.perfops.numParts                        1
 
-spark.numSplits                                   1
+spark.cdm.schema.origin.column.names              key,value,fruits
+spark.cdm.schema.origin.column.partition.names    key
+spark.cdm.schema.target.column.id.names           key,fruit
+spark.cdm.schema.origin.column.types              0,0,5%0%1
 
-spark.query.origin                                key,value,fruits
-spark.query.origin.partitionKey                   key
-spark.query.target.id                             key,fruit
-spark.query.types                                 0,0,5%0%1
+spark.cdm.feature.explodeMap.origin.name          fruits
+spark.cdm.feature.explodeMap.target.name.key      fruit
+spark.cdm.feature.explodeMap.target.name.value    fruit_qty
 
-spark.cdm.cql.feature.explodeMap.origin.name        fruits
-spark.cdm.cql.feature.explodeMap.target.name.key    fruit
-spark.cdm.cql.feature.explodeMap.target.name.value  fruit_qty
+spark.cdm.autocorrect.missing                     true
+spark.cdm.autocorrect.mismatch                    true
 
-spark.target.autocorrect.missing                  true
-spark.target.autocorrect.mismatch                 true

SIT/features/02_explode_map/migrate.properties

@@ -1,22 +1,16 @@
-spark.origin.host                                 cdm-sit-cass
-spark.origin.port                                 9042
-spark.origin.username                             cassandra
-spark.origin.password                             cassandra
-spark.origin.keyspaceTable                        source.feature_explode_map
+spark.cdm.origin.connect.host                     cdm-sit-cass
+spark.cdm.target.connect.host                     cdm-sit-cass
 
-spark.target.host                                 cdm-sit-cass
-spark.target.port                                 9042
-spark.target.username                             cassandra
-spark.target.password                             cassandra
-spark.target.keyspaceTable                        target.feature_explode_map
+spark.cdm.schema.origin.keyspaceTable             origin.feature_explode_map
+spark.cdm.schema.target.keyspaceTable             target.feature_explode_map
+spark.cdm.perfops.numParts                        1
 
-spark.numSplits                                   1
+spark.cdm.schema.origin.column.names              key,value,fruits
+spark.cdm.schema.origin.column.partition.names    key
+spark.cdm.schema.target.column.id.names           key,fruit
+spark.cdm.schema.origin.column.types              0,0,5%0%1
 
-spark.query.origin                                key,value,fruits
-spark.query.origin.partitionKey                   key
-spark.query.target.id                             key,fruit
-spark.query.types                                 0,0,5%0%1
+spark.cdm.feature.explodeMap.origin.name          fruits
+spark.cdm.feature.explodeMap.target.name.key      fruit
+spark.cdm.feature.explodeMap.target.name.value    fruit_qty
 
-spark.cdm.cql.feature.explodeMap.origin.name        fruits
-spark.cdm.cql.feature.explodeMap.target.name.key    fruit
-spark.cdm.cql.feature.explodeMap.target.name.value  fruit_qty

SIT/features/02_explode_map/setup.cql

@@ -1,8 +1,8 @@
-DROP TABLE IF EXISTS source.feature_explode_map;
-CREATE TABLE source.feature_explode_map(key text, value text, fruits map<text,int>, PRIMARY KEY (key));
-INSERT INTO source.feature_explode_map(key,value,fruits) VALUES ('key1','valueA', {'apples': 3, 'oranges': 5, 'bananas': 2, 'grapes': 11});
-INSERT INTO source.feature_explode_map(key,value,fruits) VALUES ('key2','valueB', {'apples': 4, 'oranges': 6, 'bananas': 3, 'pears': 7});
-INSERT INTO source.feature_explode_map(key,value,fruits) VALUES ('key3','valueC', {'apples': 5, 'oranges': 7, 'bananas': 4, 'kiwi': 42});
+DROP TABLE IF EXISTS origin.feature_explode_map;
+CREATE TABLE origin.feature_explode_map(key text, value text, fruits map<text,int>, PRIMARY KEY (key));
+INSERT INTO origin.feature_explode_map(key,value,fruits) VALUES ('key1','valueA', {'apples': 3, 'oranges': 5, 'bananas': 2, 'grapes': 11});
+INSERT INTO origin.feature_explode_map(key,value,fruits) VALUES ('key2','valueB', {'apples': 4, 'oranges': 6, 'bananas': 3, 'pears': 7});
+INSERT INTO origin.feature_explode_map(key,value,fruits) VALUES ('key3','valueC', {'apples': 5, 'oranges': 7, 'bananas': 4, 'kiwi': 42});
 
 DROP TABLE IF EXISTS target.feature_explode_map;
 CREATE TABLE target.feature_explode_map(key text, fruit text, value text, fruit_qty int, PRIMARY KEY ((key), fruit));