cassandra-data-migrator (also known as CDM)

Migrate and Validate Tables between Origin and Target Cassandra Clusters.

[!IMPORTANT] Please note the below listed Prerequisite to avoid CDM, Spark, Scala, Hadoop version mismatch issues

Install as a Container

• Get the latest image that includes all dependencies from DockerHub
  • All migration tools (cassandra-data-migrator + dsbulk + cqlsh) would be available in the /assets/ folder of the container

Install as a JAR file

• Download the latest jar file using one of the approaches below,
• curl -L0 https://github.com/datastax/cassandra-data-migrator/releases/download/x.y.z/cassandra-data-migrator-x.y.z.jar --output cassandra-data-migrator-x.y.z.jar (OR)
• Download from packages area here

Prerequisite CDM 6.x+

• Java17 (minimum) as Spark 4.x binaries are compiled with it.
• Any Java 17+ LTS version (e.g. Java 21 or Java 25) could be used if needed
• Spark version 4.1.2

Spark can be installed by running the following: -

wget https://archive.apache.org/dist/spark/spark-4.1.2/spark-4.1.2-bin-hadoop3.tgz
tar -xvzf spark-4.1.2-bin-hadoop3.tgz

Prerequisite CDM 5.x+

• Java11 (minimum) as Spark 3.x binaries are compiled with it.
• Any Java 11+ LTS version (e.g. Java 17, Java 21 or Java 25) could be used if needed
• Spark version 3.5.8 with Scala 2.13

Spark can be installed by running the following: -

wget https://archive.apache.org/dist/spark/spark-3.5.8/spark-3.5.8-bin-hadoop3-scala2.13.tgz 
tar -xvzf spark-3.5.8-bin-hadoop3-scala2.13.tgz 

[!CAUTION] You may see an exception like below if there are any version mismatches,

Exception in thread "main" java.lang.NoSuchMethodError: 'void scala.runtime.Statics.releaseFence()'

Note: - Typically installed using binaries mentioned above on a single VM (no cluster necessary) where you want to run this job. This simple setup is recommended for most one-time migrations. - You could use a Spark Cluster or a Spark Serverless platform like Watsonx.data or Databricks or Google Dataproc (that supports the above mentioned versions) for large (e.g. several terabytes) complex migrations OR when CDM is used as a long-term data-transfer utility and not a one-time job. - When deploying CDM on a Spark cluster, replace the params --master "local[*]" with --master "spark://master-host:port" and remove any params (e.g. --driver-memory, --executor-memory, etc.) related to a single VM run

Steps for Data-Migration:

1. cdm.properties file needs to be configured as applicable for the environment. The file can have any name, it does not need to be cdm.properties.

   • A sample properties file with default values can be found here as cdm.properties
   • A complete reference properties file with default values can be found here as cdm-detailed.properties

2. Place the properties file where it can be accessed while running the job via spark-submit.
3. Run the job using spark-submit command as shown below:

spark-submit --properties-file cdm.properties \
--conf spark.cdm.schema.origin.keyspaceTable="<keyspacename>.<tablename>" \
--master "local[*]" --driver-memory 25G --executor-memory 25G \
--class com.datastax.cdm.job.Migrate cassandra-data-migrator-5.x.x.jar &> logfile_name_$(date +%Y%m%d_%H_%M).txt

Note: - Above command generates a log file logfile_name_*.txt to avoid log output on the console. - Update the memory options (driver & executor memory) based on your use-case - To track details of a run (recorded on the target keyspace), pass param --conf spark.cdm.trackRun=true - To filter records only for a specific token range, pass the below two additional params to the Migration OR Validation job

--conf spark.cdm.filter.cassandra.partition.min=<token-range-min>
--conf spark.cdm.filter.cassandra.partition.max=<token-range-max>

Steps for Data-Validation:

• To run the job in Data validation mode, use class option --class com.datastax.cdm.job.DiffData as shown below

spark-submit --properties-file cdm.properties \
--conf spark.cdm.schema.origin.keyspaceTable="<keyspacename>.<tablename>" \
--master "local[*]" --driver-memory 25G --executor-memory 25G \
--class com.datastax.cdm.job.DiffData cassandra-data-migrator-5.x.x.jar &> logfile_name_$(date +%Y%m%d_%H_%M).txt

• Validation job will report differences as “ERRORS” in the log file as shown below.

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.
  • Note that it lists differences by primary-key values.
• If you would like to redirect such logs (rows with details of missing and mismatched rows) into a separate file, you could use the log4j2.properties file provided here as shown below

spark-submit --properties-file cdm.properties \
--conf spark.cdm.schema.origin.keyspaceTable="<keyspacename>.<tablename>" \
--conf spark.executor.extraJavaOptions='-Dlog4j2.configurationFile=log4j2.properties' \
--conf spark.driver.extraJavaOptions='-Dlog4j2.configurationFile=log4j2.properties' \
--master "local[*]" --driver-memory 25G --executor-memory 25G \
--class com.datastax.cdm.job.DiffData cassandra-data-migrator-5.x.x.jar &> logfile_name_$(date +%Y%m%d_%H_%M).txt

• The Validation job can also be run in an AutoCorrect mode. This mode can
  • Add any missing records from origin to target
  • Update any mismatched records between origin and target
• Enable/disable this feature using one or both of the below params in the properties file

spark.cdm.autocorrect.missing                     false|true
spark.cdm.autocorrect.mismatch                    false|true

[!IMPORTANT] The validation job will never delete records from target i.e. it only adds or updates data on target

Rerun (previously incomplete) Migration or Validation

• You can rerun/resume the last Migration or Validation job that stopped (or completed with some errors) for any reasons by setting spark.cdm.trackRun.autoRerun to true (default is false) as shown below. This will auto-discover the progress of the last job and will resume from that point i.e. it will skip any token-ranges from the previous run that were migrated (or validated) successfully.

spark-submit --properties-file cdm.properties \
 --conf spark.cdm.schema.origin.keyspaceTable="<keyspacename>.<tablename>" \
 --conf spark.cdm.trackRun.autoRerun=true \
 --master "local[*]" --driver-memory 25G --executor-memory 25G \
 --class com.datastax.cdm.job.<Migrate|DiffData> cassandra-data-migrator-5.x.x.jar &> logfile_name_$(date +%Y%m%d_%H_%M).txt

• You can also resume a specific earlier job (not the last one) that stopped (or completed with some errors) for any reasons by passing a specific spark.cdm.trackRun.previousRunId as shown below

spark-submit --properties-file cdm.properties \
 --conf spark.cdm.schema.origin.keyspaceTable="<keyspacename>.<tablename>" \
 --conf spark.cdm.trackRun.previousRunId=<prev_run_id> \
 --master "local[*]" --driver-memory 25G --executor-memory 25G \
 --class com.datastax.cdm.job.<Migrate|DiffData> cassandra-data-migrator-5.x.x.jar &> logfile_name_$(date +%Y%m%d_%H_%M).txt

Perform large-field Guardrail violation checks

• This mode can help identify large fields on an origin table that may break you cluster guardrails (e.g. AstraDB has a 10MB limit for a single large field), use class option --class com.datastax.cdm.job.GuardrailCheck as shown below

spark-submit --properties-file cdm.properties \
--conf spark.cdm.schema.origin.keyspaceTable="<keyspacename>.<tablename>" \
--conf spark.cdm.feature.guardrail.colSizeInKB=10000 \
--master "local[*]" --driver-memory 25G --executor-memory 25G \
--class com.datastax.cdm.job.GuardrailCheck cassandra-data-migrator-5.x.x.jar &> logfile_name_$(date +%Y%m%d_%H_%M).txt

[!NOTE] This mode only operates on one database i.e. origin, there is no target in this mode

Features

• Auto-detects table schema (column names, types, keys, collections, UDTs, etc.)
  • Including counter table Counter tables
• Rerun/Resume a previous job that may have stopped for any reason (killed, had exceptions, etc.)
  • If you rerun a validation job, it will include any token-ranges that had differences in the previous run
  • If reruns are repeatedly failing, you can use the rerunMultiplier feature to help improve the chances of success.
• Preserve writetimes and TTLs
• Supports migration/validation of advanced DataTypes (Sets, Lists, Maps, UDTs)
• Filter records from Origin using writetime and/or CQL conditions and/or a list of token-ranges
• Perform guardrail checks (identify large fields)
• Supports adding constants as new columns on Target
• Supports expanding Map columns on Origin into multiple records on Target
• Supports extracting value from a JSON column in Origin and map it to a specific field on Target
• Can be deployed on a Spark Cluster or a single VM
• Fully containerized (Docker and K8s friendly)
• SSL Support (including custom cipher algorithms)
• Migrate from any Cassandra Origin (Apache Cassandra® / DataStax Enterprise™ / DataStax Astra DB™) to any Cassandra Target (Apache Cassandra® / DataStax Enterprise™ / DataStax Astra DB™)
• Automatic download of Secure Connect Bundles for Astra DB using the DevOps API
• Supports migration/validation from and to Azure Cosmos Cassandra
• Validate migration accuracy and performance using a smaller randomized data-set
• Supports adding custom fixed writetime and/or ttl
• Track run information (start-time, end-time, run-metrics, status, etc.) in tables (cdm_run_info and cdm_run_details) on the target keyspace

Things to know

[!TIP] If you want to pass in additional Cassandra Java Driver configs, you can pass in --conf spark.driver.extraJavaOptions="-Ddatastax-java-driver.advanced.connection.pool.remote.size=5

[!TIP] If you want to log DEBUG level statements for troubleshooting, you can pass in --conf spark.driver.extraJavaOptions='-Dlog4j2.level=DEBUG -Dlog4j2.rootLogger.level=DEBUG' --conf spark.executor.extraJavaOptions='-Dlog4j2.level=DEBUG -Dlog4j2.rootLogger.level=DEBUG'

• Each run (Migration or Validation) can be tracked (when enabled). You can find summary and details of the same in tables cdm_run_info and cdm_run_details in the target keyspace.
• CDM does not migrate ttl & writetime at the field-level (for optimization reasons). It instead finds the field with the highest ttl & the field with the highest writetime within an origin row and uses those values on the entire target row.
• CDM ignores using collection and UDT fields for ttl & writetime calculations by default for performance reasons. If you want to include such fields, set spark.cdm.schema.ttlwritetime.calc.useCollections param to true.
• If a table has only collection and/or UDT non-key columns and no