In order to connect to Amazon Keyspaces and the Cassandra cluster you need configure reference
configuration files ../glue/conf/CassandraConnector.conf and ../glue/conf/KeyspacesConnector.conf
in ../conf accordingly to the environment.
The following command initializes the CQLReplicator environment, which involves the copying JAR artifacts, creation a Glue connector, a S3 bucket, a Glue job, migration keyspace, and ledger table.
cd cql-replicator/glue/bin &&
cqlreplicator --cmd init --sg '"sg-1","sg-2"' \
--subnet "subnet-XXXXXXXXXXXX" --az us-east-1a --region us-east-1 \
--env ver5 \ # Optional
--glue-iam-role glue-cassandra-migration \
--landing-zone s3://<PROVIDED_BUCKET_NAME> \
--target-type keyspacesAfter running the below command, you should find in the AWS account:
- CQLReplicator-ver5 glue job and Glue connector in AWS Glue console
- Amazon S3 bucket that stores the artifacts
- Keyspace
migrationandledgertable in Amazon Keyspaces
Note: If you run the CQLReplicator on EC2 you should attach the following inline policy:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"iam:GetContextKeysForPrincipalPolicy",
"iam:SimulatePrincipalPolicy",
"iam:ListAttachedRolePolicies",
"iam:PassRole",
"glue:*",
"pricing:GetProducts",
"pricing:DescribeServices",
"pricing:GetAttributeValues"
],
"Resource": "*"
}
]
}--sg- these security group(s) which allow access to the Cassandra cluster from AWS Glue and includes the self-referencing inbound rule for all traffic--subnet- this is the subnet to which the Cassandra cluster belongs. In AWS Glue, when a Glue job needs to run in your Virtual Private Cloud (VPC), AWS Glue sets up elastic network interfaces that enable your Glue jobs to connect securely to other resource within your VPC. Each elastic network interface is assigned a private IP address from the IP address range within the subnet you specified. For example, 20 workers need 20 IP addresses.--az- this is the AZ of the subnet--region- AWS region where the Cassandra cluster is deployed--glue-iam-role- you need the IAM role permissions that AWS Glue can assume when calling other services on your behalf. Services: Amazon Keyspaces and S3--landing-zone- this is an optional parameter, you can reuse any existing S3 buckets in the AWS account If you don't supply it, init process will try to create a new bucket to store config, JAR artifacts, and intermediate files--target-type- this is an optional parameter, the default value iskeyspaces, available options:parquet,memorydb,andopensearch--skip-glue-connector- this is an optional parameter allows you to skip the glue connector creation step--skip-keyspaces-ledger- in case if you don't want to create/delete the ledger (the internal table in Amazon Keyspaces migration.ledger)--main-script-landing- use this flag if you want to store the main script in thes3://your-landing-zone/script/CQLReplicator.scala--env- use to create multiple environments, e.g. dev/test/prod
CQLReplicator uses Apache Iceberg with the AWS Glue Data Catalog for snapshot management. The Glue job runs inside a VPC, so the following VPC endpoints are required for Iceberg catalog operations:
| Endpoint | Type | Purpose |
|---|---|---|
com.amazonaws.{region}.glue |
Interface | Iceberg catalog operations (CREATE TABLE, getTable, etc.) |
com.amazonaws.{region}.s3 |
Gateway | Iceberg data/metadata read/write in S3 |
The Glue interface endpoint must have:
PrivateDnsEnabled=true- The VPC must have
enableDnsSupport=trueandenableDnsHostnames=true - The endpoint's security group must allow inbound TCP port 443 from the Glue job's security group
Verification:
# Check Glue endpoint
aws ec2 describe-vpc-endpoints \
--filters "Name=vpc-id,Values={vpc-id}" "Name=service-name,Values=com.amazonaws.{region}.glue" \
--query 'VpcEndpoints[*].{ID:VpcEndpointId,Subnets:SubnetIds,State:State,PrivateDns:PrivateDnsEnabled}'
# Check S3 endpoint
aws ec2 describe-vpc-endpoints \
--filters "Name=vpc-id,Values={vpc-id}" "Name=service-name,Values=com.amazonaws.{region}.s3" \
--query 'VpcEndpoints[*].{ID:VpcEndpointId,State:State}'To operate CQLReplicator on AWS Glue, you need to use --cmd run command, followed by a series of parameters. The
precise configuration of these is primarily determined by your unique migration requirements. For instance, these
settings may vary if you choose to replicate TTLs, updates, or offloading objects exceeding 1MB to Amazon S3.
The CQLReplicator offers three options for operation:
- For high-volume traffic exceeding 5000 Write Capacity Units (WCU) per second:
- Use the command:
--max-wcu-traffic 20000 - This setting uses the default of 4 tiles (Glue jobs)
- Use the command:
- For lower-volume traffic below 5000 WCU per second:
- Use the command:
--max-wcu-traffic 4000 --tiles 1 - This setting customizes the number of tiles to 1
- Use the command:
- For custom configurations:
- Use the command: --tiles 12 (without specifying --max-wcu-traffic)
- This allows you to set a custom number of tiles with your preferred worker type
- Note: Changing the number of tiles requires to resize the migration process
- When estimating tile requirements:
- Each tile (concurrent Glue job) can process up to 74 million primary keys
- This is achieved using 299 DPUs (G.025X) maximum
- For example, 12 tiles can handle up to 1 billion primary keys
These guidelines help optimize the CQLReplicator's performance based on your specific traffic and data volume needs.
Let's run the following command to replicate the workload from the Cassandra cluster to Amazon Keyspaces with the write traffic 20000 WCUs per second:
Note: The following command is going to capture only "insert" or "delete". In order, to capture "update" you need to add an arbitrary column from you table
that is constantly updated, for example, add --writetime-column your-column to the following command line.
cqlreplicator --cmd run --max-wcu-traffic 20000 --landing-zone s3://<PROVIDED_BUCKET_NAME> --region us-east-1 \
--src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator \
--env ver5 --target-type keyspaces
The write traffic against the Amazon Keyspaces table should not exceed 20000 WCUs per second
Your source keyspace and table ks_test_cql_replicator.test_cql_replicator in the Cassandra cluster.
Your target keyspace and table ks_test_cql_replicator.test_cql_replicator in Amazon Keyspaces.
Parameter --inc-traffic - enables incremental traffic to not overload the Cassandra Cluster and Amazon Keyspaces
with high number of requests. CQLReplicator is going to increase traffic every 4 minutes. In the following
example with 4 tiles and one cassandra row is under 2KB, the traffic is going to be gradually increased every 4
minutes. At peak traffic, it can reach up to 6,300 WCUs per second.
cqlreplicator --cmd run --max-wcu-traffic 6000 --landing-zone s3://<PROVIDED_BUCKET_NAME> --region us-east-1 \
--writetime-column col3 --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator \
--env ver5 --inc-traffic --target-type keyspaces
The --writetime-column parameter accepts multiple comma-separated column names. When multiple columns are specified,
CQLReplicator computes the maximum WRITETIME() across all specified columns to detect the most recent mutation for change detection.
Single column (existing behavior):
cqlreplicator --cmd run --max-wcu-traffic 20000 --landing-zone s3://<PROVIDED_BUCKET_NAME> --region us-east-1 \
--writetime-column col1 --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator \
--env ver5 --target-type keyspacesMultiple columns:
cqlreplicator --cmd run --max-wcu-traffic 20000 --landing-zone s3://<PROVIDED_BUCKET_NAME> --region us-east-1 \
--writetime-column col1,col2 --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator \
--env ver5 --target-type keyspacesWhen multiple columns are specified, CQLReplicator generates greatest(writetime(col1), writetime(col2), ...) as ts
in the CQL query. Cassandra's WRITETIME() returns null for columns that were never explicitly written. The greatest()
function excludes null values and returns the maximum of the non-null writetimes. If all specified columns have null
writetimes for a row, the result is null.
By default, CQLReplicator uses UNLOGGED BATCH operations for improved performance and cost efficiency. However, when dealing with large partitions, you may want to disable batching and use single inserts instead to avoid throttling issues in Amazon Keyspaces. This can help ensure smoother migration of large partitions.
cqlreplicator --cmd run --max-wcu-traffic 6000 --landing-zone s3://<PROVIDED_BUCKET_NAME> --region us-east-1 \
--writetime-column col3 --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator \
--env ver5 --inc-traffic --target-type keyspaces --json-mapping '{"keyspaces": {"writeConfiguration": { "maxStatementsPerBatch": 1}}}’If you want to replicate only the historical workload and release Glue resources once it's completed, add the following flag
--workload-type batch to the command line, for example:
cqlreplicator --cmd run --max-wcu-traffic 20000 --landing-zone s3://<PROVIDED_BUCKET_NAME> --region us-east-1 \
--writetime-column col3 --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator \
--env ver5 --inc-traffic --target-type keyspaces --env ver5 --workload-type batchthe TTL feature should be enabled on the target table before running the following command
cqlreplicator --cmd run --max-wcu-traffic 20000 --landing-zone s3://<PROVIDED_BUCKET_NAME> --region us-east-1 \
--writetime-column col3 --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator --ttl-column col3 \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator --inc-traffic --env ver5 --target-type keyspacesthe Client-Side Timestamps feature should be enabled on the target table before running the following command
cqlreplicator --cmd run --max-wcu-traffic 20000 --landing-zone s3://<PROVIDED_BUCKET_NAME> --region us-east-1 \
--writetime-column col3 --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator --env ver5 \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator \
--target-type keyspaces --json-mapping '
{
"replication": {
"replicateWithTimestamp": true
}
}'Before running the migration process configure lifecycle for the S3 bucket to delete objects after expiration. The xref column should be present in the source and the target tables before running the cqlreplicator.
The payload will be compressed by LZ4 and stored in the S3 bucket. The compressed payload includes the length of the original decompressed data, so that you don't need to save the length at anywhere else. The compressed data must be decompressed by LZ4DecompressorWithLength and is NOT compatible with any other decompressors in lz4-java or any other lz4 tools. This class deliberately does not extend LZ4Compressor because they are not interchangeable.
./cqlreplicator --cmd run --max-wcu-traffic 20000 --landing-zone s3://<PROVIDED_BUCKET_NAME> --src-keyspace ks_test_cql_replicator \
--src-table test_cql_replicator --trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator --env ver5 \
--region us-east-1 --ttl-column col3 --target-type keyspaces --json-mapping '
{
"replication": {
"allColumns": true,
"columns":[""],
"useCustomSerializer": false,
"useMaterializedView": {
"enabled": false,
"mvName": "mvName"
}
},
"keyspaces": {
"compressionConfig": {
"enabled": false,
"compressAllNonPrimaryColumns": false,
"compressNonPrimaryColumns": [""],
"targetNameColumn": ""
},
"largeObjectsConfig": {
"enabled": true,
"column": "col1",
"bucket": "my-application-resource",
"prefix": "ks_test_cql_replicator/test_cql_replicator/col1",
"enableRefByTimeUUID": false,
"xref": "link"
}
}}'If you enable enableRefByTimeUUID the S3 bucket will be populated in the following format s3://<PROVIDED_BUCKET_NAME>/ks_test_cql_replicator/test_cql_replicator/col1/.
The target column link will be populated by a referencing timeUUID, and col1 will be skipped.
If you disable enableRefByTimeUUID the S3 bucket will be populated in the following format s3://<PROVIDED_BUCKET_NAME>/ks_test_cql_replicator/test_cql_replicator/col1/key=:/payload,
where col0 is the clustering column, and key is the partition key. The target column link and col1 will be skipped.
When the CQLReplicator encounters persistent failures during replication, failed rows are automatically stored in a Dead Letter Queue (DLQ) directory within your S3 bucket. Failed rows are stored in the following S3 path format: s3://<PROVIDED_BUCKET_NAME>/ks-name/tbl-name/dlq.
Common failure scenarios:
- When source partitions exceed tens of megabytes
- When operations exceed the maximum retry limit (64 attempts). Usually indicates persistent connectivity or resource issues
- When the target table schema doesn't match the source. Fix the schema before replaying
Best practises:
- Review the failed rows in DLQ to understand the failure patterns
- Address any underlying issues (schema, connectivity, resources) before replay
- Monitor the replay process to ensure successful completion
To replay failed rows:
- Stop the CQLReplicator with
request-stop - Run it again with
--replay-log- this option allows you to replay failed operations during the replication phase.
To stop migration process gracefully run the following command for all tiles:
cqlreplicator --cmd request-stop --landing-zone s3://<PROVIDED_BUCKET_NAME> --region us-east-1-ver5 \
--src-keyspace ks_test_cql_replicator --src-table test_cql_replicator--target-type keyspaces --env ver5To stop migration process gracefully run the following command for a single tile:
cqlreplicator --cmd request-single-stop --tile 1 --landing-zone s3://<PROVIDED_BUCKET_NAME> --region us-east-1 \
--src-keyspace ks_test_cql_replicator --src-table test_cql_replicator --target-type keyspaces --env ver5To start migration process for a single tile:
cqlreplicator --cmd run-single --tile 1 --tiles 4 --landing-zone s3://<PROVIDED_BUCKET_NAME> --region us-east-1 \
--src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator --inc-traffic --target-type keyspaces --env ver5To force an immediate termination of the CQLReplicator:
cqlreplicator --cmd kill --landing-zone s3://<PROVIDED_BUCKET_NAME> --region us-east-1 \
--src-keyspace ks_test_cql_replicator --src-table test_cql_replicator --target-type keyspaces --env ver5Customers can simply restart the migration process from the point where it was interrupted by restarting the failed
CQLReplicators jobs. To restart the failed CQLReplicators jobs, you need to rerun your run command with the same
parameters
if both the discovery and some of the replicators have failed. If the discovery job is still running and some of the
replicators have failed, you need to use extra flag --skip-discovery.
If you wish to reset the CQLReplicator for a specific table, add the --cleanup-requested or --cr flag to your run command.
This action will clean up both the internal CQLReplicator ledger table and the S3 data, but don't forget to remove it
if you want to stop it and run it again.
To obtain the number of replicated rows during the back filling phase, run the following command:
cqlreplicator --cmd stats --landing-zone s3://<PROVIDED_BUCKET_NAME> \
--src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--target-type keyspaces --region us-east-1 --env ver5To obtain the number of replicated rows after the back filling phase, run the following command:
cqlreplicator --cmd stats --landing-zone s3://<PROVIDED_BUCKET_NAME> \
--src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--target-type keyspaces --region us-east-1 --replication-stats-enabled --env ver5 ___ ___ _ ____ _ _ _
/ ___/ _ \| | | _ \ ___ _ __ | (_) ___ __ _| |_ ___ _ __
| | | | | | | | |_) / _ \ '_ \| | |/ __/ _` | __/ _ \| '__|
| |__| |_| | |___| _ < __/ |_) | | | (_| (_| | || (_) | |
\____\__\_\_____|_| \_\___| .__/|_|_|\___\__,_|\__\___/|_|
|_|
·······································································
: __ _______ _____ _____ :
: /\ \ / / ____| | __ \ / ____| :
: / \ \ /\ / / (___ | |__) | __ ___| (___ ___ _ ____ _____ :
: / /\ \ \/ \/ / \___ \ | ___/ '__/ _ \\___ \ / _ \ '__\ \ / / _ \:
: / ____ \ /\ / ____) | | | | | | (_) |___) | __/ | \ V / __/:
:/_/ \_\/ \/ |_____/ |_| |_| \___/_____/ \___|_| \_/ \___|:
·······································································
[2024-02-02T16:50:18-05:00] OS: Linux
+------------------------------------------------------------------------+
| Tile | Inserts | Updates | Deletes | Timestamp |
+------------------------------------------------------------------------+
| 0 | 4 | 1 | 0 | "2024-02-02T21:50:12.888" |
+------------------------------------------------------------------------+
| 1 | 2 | 2 | 1 | "2024-02-02T21:50:25.454" |
+------------------------------------------------------------------------+
[2024-02-02T16:50:39-05:00] Discovered rows in ks_test_cql_replicator.test_cql_replicator is 70,270
[2024-02-02T16:50:39-05:00] Replicated rows in ks_test_cql_replicator.test_cql_replicator is 70,270
To retrieve errors from Glue jobs, you can execute the following command:
cqlreplicator --cmd get-failed-jobs --landing-zone s3://<PROVIDED_BUCKET_NAME> \
--region us-east-1 --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicatorBy default, CQLReplicator sets the Spark log level to ERROR to reduce noise in CloudWatch logs. You can change this using the --logging-level flag:
cqlreplicator --cmd run --max-wcu-traffic 20000 --landing-zone s3://<PROVIDED_BUCKET_NAME> --region us-east-1 \
--src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator \
--env ver5 --target-type keyspaces --logging-level WARNAvailable log levels: ALL, TRACE, DEBUG, INFO, WARN, ERROR (default), FATAL, OFF.
This controls Spark's internal logging (DAGScheduler, BlockManager, TaskSetManager, etc.). CQLReplicator's own messages via GlueLogger are always visible regardless of this setting.
You can profile and monitor CQLReplicator operations using AWS Glue job profiler. It collects and processes raw data from AWS Glue jobs into readable, near real-time metrics stored in Amazon CloudWatch. These statistics are retained and aggregated in CloudWatch so that you can access historical information for a better perspective on how your application is performing.
In order tp enable the CQLReplicator enhanced monitoring use the --enhanced-monitoring-enabled flag with your init
command.
You may incur additional charges when you enable job metrics and CloudWatch custom metrics are created
if you want to replicate your data from a timestamp simply add the flag --start-replication-from followed by
the timestamp to your command line:
cqlreplicator --cmd --max-wcu-traffic 20000 --landing-zone s3://<PROVIDED_BUCKET_NAME> \
--writetime-column col1 --region us-east-1 \
--src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator \
--target-type keyspaces --start-replication-from 1706408365278858 --env ver5if you want to change the default point in time replication behavior and replicate data before timestamp then specify predicate in json mapping:
{
"replication": {
"pointInTimeReplicationConfig": {
"predicateOp": "lessThanOrEqual"
}
}
}To minimize the size of large columns in the target table, enable the compressionConfig in the json mapping.
All selected columns will be encoded in JSON format and compressed by the LZ4 fast compressor.
Note that the compressed payload includes the length of the original decompressed data, so that you don't need to save the length at anywhere else. The compressed data must be decompressed by LZ4DecompressorWithLength and is NOT compatible with any other decompressors in lz4-java or any other lz4 tools. This class deliberately does not extend LZ4Compressor because they are not interchangeable.
./cqlreplicator --cmd run --max-wcu-traffic 20000 --landing-zone s3://<PROVIDED_BUCKET_NAME> --writetime-column col2 \
--region us-east-1 --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator_compressed_columns --env ver5 \
--target-type keyspaces --json-mapping '{
"replication": {
"allColumns": true,
"columns":[],
"useCustomSerializer": false
},
"keyspaces": {
"compressionConfig": {
"enabled": true,
"compressAllNonPrimaryColumns": false,
"compressNonPrimaryColumns": ["col2","col3","col4","col5","col6","col7","col8"],
"targetNameColumn": "compressedPayload"
}}
}'if you want to change the replicated columns set allColumns:false, and set the columns
list columns:["col1","col2"]
in the json mapping during the run, for example,
./cqlreplicator --cmd run --max-wcu-traffic 20000 --landing-zone s3://<PROVIDED_BUCKET_NAME>--writetime-column col2 \
--region us-east-1 --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator_compressed_columns --env ver5 \
--target-type keyspaces --json-mapping '{
"replication": {
"allColumns": false,
"columns":["col1", "col2"],
"useCustomSerializer": false
},
"keyspaces": {
"compressionConfig": {
"enabled": false,
"compressAllNonPrimaryColumns": false,
"compressNonPrimaryColumns": ["col2","col3","col4","col5","col6","col7","col8"],
"targetNameColumn": "compressedPayload"
}}
}'./cqlreplicator --cmd run --max-wcu-traffic 20000 --landing-zone s3://<PROVIDED_BUCKET_NAME> --writetime-column col2 \
--region us-east-1 --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator_filtered_columns --env ver5 \
--target-type keyspaces --json-mapping '{
"replication": {
"allColumns": false,
"columns":["key","col0","col1","col2","col3","col4","col5","col6"],
"useCustomSerializer": false
},
"keyspaces": {
"compressionConfig": {
"enabled": false,
"compressAllNonPrimaryColumns": false,
"compressNonPrimaryColumns": ["payload"],
"targetNameColumn": "compressed"
}}
}'There are use-cases when you need to filter out some certain keys during the migration phase, for example:
./cqlreplicator --cmd run --max-wcu-traffic 20000 --landing-zone s3://<PROVIDED_BUCKET_NAME> --writetime-column col2 \
--region us-east-1 --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator_filtered_columns --env ver5 \
--target-type keyspaces --json-mapping '{
"keyspaces": {
"transformation": {
"enabled": true,
"filterExpression": "client_id NOT like \"ACME%\""
}
}
}'./cqlreplicator --cmd run --max-wcu-traffic 20000 --landing-zone s3://<PROVIDED_BUCKET_NAME> --writetime-column col2 \
--region us-east-1 --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator_filtered_columns --env ver5 \
--target-type keyspaces --json-mapping '{
"keyspaces": {
"transformation": {
"enabled": true,
"addNonPrimaryKeyColumns": ["col2"],
"filterExpression": "col2 > timestamp \"2024-11-12 00:00:59.588000+0000\""
}
}
}'Note: You can use only primary key columns in the filterExpression and it should be defined as the Spark SQL Expression.
CQLReplicator supports data transformation during migration using various hash functions. This feature is useful for:
- Data anonymization: Hash sensitive data like emails, phone numbers, or personal identifiers
- Data obfuscation: Transform data while maintaining referential integrity
- Compliance requirements: Meet data privacy regulations by hashing PII data
- MurmurHash3: Fast, non-cryptographic hash function (returns integer)
- MD5: 128-bit cryptographic hash (returns hex string)
- SHA-1: 160-bit cryptographic hash (returns hex string)
- SHA-256: 256-bit cryptographic hash (returns hex string)
- XXHASH64: Fast, 64-bit non-cryptographic hash (returns long integer)
columnName: The source column to transform to hash valuerule: Hash function to apply (murmurhash3, md5, sha-1, sha-256, xxhash64)alias: Optional new column name (if not provided, replaces original column)keepSource: Boolean to keep original column when using alias (default: false)
cqlreplicator --cmd run --max-wcu-traffic 20000 --landing-zone s3://<BUCKET_NAME> \
--region us-east-1 --src-keyspace source_ks --src-table users \
--trg-keyspace target_ks --trg-table users_transformed --env prod \
--target-type keyspaces --json-mapping '{
"keyspaces": {
"transformation": {
"enabled": true,
"transformExpressions": [
{
"columnName": "email",
"rule": "md5",
"alias": "email_hash",
"keepSource": false
},
{
"columnName": "phone",
"rule": "sha-256",
"alias": "phone_hash",
"keepSource": true
},
{
"columnName": "user_id",
"rule": "murmurhash3"
}
]
}
}
}'This example:
- Hashes
emailto MD5 and replaces it withemail_hash - Hashes
phoneto SHA-256 asphone_hashwhile keeping originalphonecolumn - Hashes
user_idin place using MurmurHash3
In order to preserve existing records during the bulk migration phase set readBeforeWrite to true in the json-mapping, for example:
./cqlreplicator --cmd run --max-wcu-traffic 20000 --landing-zone s3://<PROVIDED_BUCKET_NAME> --writetime-column col2 \
--region us-east-1 --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator_filtered_columns --env ver5 \
--target-type keyspaces --json-mapping '{
"keyspaces": {
"readBeforeWrite": true
}
}'This feature allows the CQLReplicator tool to automatically convert UDTs in C* to plain JSON strings. This enhancement provides a seamless way to transform UDTs into a more universal compatible JSON format, facilitating easier integration with other NoSQL databases.
./cqlreplicator --cmd run --max-wcu-traffic 20000 --landing-zone s3://<PROVIDED_BUCKET_NAME> --writetime-column col2 \
--region us-east-1 --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator_filtered_columns --env ver5 \
--target-type keyspaces --json-mapping '{
"keyspaces": {
"udtConversion": { "enabled": true,"columns": ["device", "user"] }
}}'By default, AWS Glue converts VARINT columns to decimal(38,0) during the read operation.
This causes values exceeding 38 digits to be implicitly converted to null, resulting in data loss during migration.
This is a known limitation of Spark's DataFrame API when reading from Cassandra.
Enable the custom VARINT reader when:
- Your source table contains VARINT columns with values exceeding 38 digits
- You notice VARINT values being converted to
nullin the target table - You need to preserve the full precision of arbitrary-precision integers
- You're migrating financial data, cryptographic keys, or other data requiring exact large integer values
The custom VARINT reader uses Cassandra's low-level RDD API instead of Spark's DataFrame API. This bypasses Spark's automatic type conversion and reads VARINT values directly as strings, preserving their full precision regardless of size.
Enable the custom VARINT reader by setting useCustomVarintReader to true in the readConfiguration:
./cqlreplicator --cmd run --max-wcu-traffic 20000 --landing-zone s3://<PROVIDED_BUCKET_NAME> \
--region us-east-1 --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator --env ver5 \
--writetime-column col2 --target-type keyspaces --json-mapping '{
"replication": {
"readConfiguration": {
"useCustomVarintReader": true,
"concurrentReads": 32,
"consistencyLevel": "LOCAL_ONE",
"fetchSizeInRows": 500
}
}
}'The readConfiguration section supports the following parameters:
useCustomVarintReader(boolean, default:false) - Enable custom VARINT handling for values over 38 digits
- The custom VARINT reader uses the RDD API which may have slightly different performance characteristics than the standard DataFrame reader
- For tables without large VARINT values, the standard reader (default) is recommended
- The custom reader is only necessary when VARINT values exceed 38 digits
The CQLReplicator relies on Cassandra JSON encoding/decoding that has been introduced in Cassandra 2.2, but if you use older Cassandra version you must enable the custom JSON Serialization on the client side:
./cqlreplicator --cmd run --max-wcu-traffic 20000 --landing-zone s3://<PROVIDED_BUCKET_NAME> --writetime-column col2 \
--region us-east-1 --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator_filtered_columns --env ver5 \
--target-type keyspaces --json-mapping '{
"replication": {
"allColumns": false,
"columns":["key","col0","col1","col2","col3","col4","col5","col6"],
"useCustomSerializer": true
},
"keyspaces": {
"compressionConfig": {
"enabled": false,
"compressAllNonPrimaryColumns": false,
"compressNonPrimaryColumns": ["payload"],
"targetNameColumn": "compressed"
}}
}'The custom JSON serializer supports the following data types: boolean, int, text, ascii, bigint, blob, counter, date, decimal, double,float,timestamp,smallint,timeuuid,uuid, and tinyint.
The following parameters must be
set allColumns:false, columns : ["partitionKey", "clusteringKey","regularColumn"],
and useCustomSerializer:true.
In order to reduce AWS Glue costs after the historical workload moved to the target storage:
- Stop the migration process:
cqlreplicator --cmd request-stop --landing-zone s3://<PROVIDED_BUCKET_NAME> --region us-east-1 \ --target-type keyspaces --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator --env ver5
- Restart the migration process with the same number of tiles but with lower number of
DPUs:
cqlreplicator --cmd run --max-wcu-traffic 5000 --landing-zone s3://<PROVIDED_BUCKET_NAME> --region us-east-1 \ --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \ --trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator --target-type keyspaces --env ver5
-
Stop the migration process:
cqlreplicator --cmd request-stop --landing-zone s3://<PROVIDED_BUCKET_NAME> --region us-east-1 \ --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator --target-type keyspaces --env ver5
-
Schedule the migration process to run it every 30 minutes:
./cqlreplicator --cmd scheduled-run --max-wcu-traffic 5000 --landing-zone s3://<PROVIDED_BUCKET_NAME> --region us-east-1 \ --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \ --trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator \ --env ver5 --target-type keyspaces --schedule "*/30 * * * *"
Note: Use only EC2 or ECS to run scheduled jobs
- Monitor costs by reviewing the automated estimation displayed when CQLReplicator initializes, which provides output such as:
The total cost of running CQLReplicator is XX.YY USD per hour - Use Cost Dashboard and AWS Cost Explorer to calculate and track cost savings from optimization efforts
To modify the number of tiles (Glue jobs) during the replication process:
-
Safely stop the current migration process: Do not force-stop any running jobs to prevent data inconsistency,
-
Ensure system state: Verify all Glue jobs show "COMPLETED" status and Check that no replication processes are actively running
-
After confirmation of complete shutdown, proceed with your desired tile count modifications using the appropriate commands:
a. Reduce the number of tiles:
./cqlreplicator --cmd resize --tiles 4 --landing-zone s3://<PROVIDED_BUCKET_NAME> --region us-east-1 \ --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator --target-type keyspaces --env ver5
Increase the number of tiles:
./cqlreplicator --cmd resize --tiles 12 --landing-zone s3://<PROVIDED_BUCKET_NAME> --region us-east-1 \ --src-keyspace ks_test_cql_replicator --src-table test_cql_replicator --target-type keyspaces --env ver5
b. After the resizing job is completed you can start the CQLReplicator again with the new number of tiles
--tiles 12
After confirming the resizing job has successfully completed, restart CQLReplicator with the updated number of tiles. Before restarting, verify that all system components are in a stable state.
| Error Message/Issue | CW Discovery logs | CW Replication logs | Work around | ||
|---|---|---|---|---|---|
| 1. No space left on device | Yes | No | Scale out the replication process by increasing the number of DPUs per discovery Glue job, for example, add --odw 15 |
||
| 2. No space left on device | Yes | No | Increase the --max-wcu-traffic |
||
| 3. You have limited IPs in the subnet | Yes | Yes | Create a new Glue connection with the new subnet and attach the CQLReplicator Glue Job | ||
| 4. Cassandra timeout during read query at consistency ONE (1 responses were required but only 0 replica responded) |
Yes | No | Reduce page size and increase the request timeout in CassandraConnector.conf | ||
| 5. Cassandra timeout during read query at consistency ONE (1 responses were required but only 0 replica responded) |
Yes | No | Create a materialized view in the Cassandra cluster with the same primary keys and one regular column. Configure the CQLReplicator to use this materialized view as the source for the primary keys. |
||
| 6. Cassandra timeout during read query at consistency ONE (1 responses were required but only 0 replica responded) |
Yes | No | Increase read_request_timeout_in_ms, range_request_timeout_in_ms, and request_timeout_in_ms in the cassandra.yaml and run rolling restart of the Cassandra cluster. |
||
| 7. java.lang.OutOfMemoryError: GC overhead limit exceeded | Yes | No | Increase the number of DPUs for the discovery job by using the parameter, e.g. --override-discovery-workers 12 |
||
| 8. Improve update/insert latency after the historical phase | No | No | Open the Glue job -> Job Details -> Advanced properties -> Job parameters, and in the Value of the key --conf reduce parameters for spark.shuffle.partitions=100 and spark.default.parallelism=100 or less |
||
| 9. The requested number value being inserted exceeds the maximum allowed precision of 38 decimal digits | No | Yes | Add the following --json-mapping '{"replication": {"readConfiguration": {"useCustomVarintReader": true}} to enable custom VARINT handling for values over 38 digits |
Work around for Cassandra timeout during read query at consistency ONE with Cassandra materialized view
- Create the materialized view
ks_test_cql_replicator_mvbased on theks_test_cql_replicatorin the Cassandra cluster with the same primary key and only one regular column - Prepare the json mapping with
"useMaterializedView": {"enabled": true, "mvName": "ks_test_cql_replicator_mv"} - Add the json mapping to the run command, e.g.
./cqlreplicator --cmd run --max-wcu-traffic 20000 --landing-zone s3://<PROVIDED_BUCKET_NAME> --writetime-column col2 \
--ttl-column reg_column --region us-east-1 --src-keyspace ks_test_cql_replicator \
--src-table ks_test_cql_replicator --trg-keyspace ks_test_cql_replicator --env ver5 \
--trg-table ks_test_cql_replicator --target-type keyspaces --json-mapping '{
"replication": {
"allColumns": true,
"columns":[""],
"useCustomSerializer": false,
"useMaterializedView":
{
"enabled": true,
"mvName": "ks_test_cql_replicator_mv"
}
},
"keyspaces": {
"compressionConfig": {
"enabled": false,
"compressAllNonPrimaryColumns": false,
"compressNonPrimaryColumns": [""],
"targetNameColumn": ""
}
}}'The following command will delete the Glue job, connector, the S3 bucket, and the ledger table:
cqlreplicator --cmd cleanup --landing-zone s3://<PROVIDED_BUCKET_NAME> --region "us-east-1"To also clean up Iceberg databases and tables from the Glue Data Catalog, provide the source keyspaces used during migration:
cqlreplicator --cmd cleanup --landing-zone s3://<PROVIDED_BUCKET_NAME> --region "us-east-1" \
--src-keyspace-list "ks_source_1,ks_source_2"Each keyspace maps to a Glue database named {keyspace}_db. Without --src-keyspace-list, the Glue Data Catalog databases are not deleted and a warning is displayed.
The reconciliation job is an on-demand CQLReplicator job that compares the
primary-key sets of a source Cassandra table and a target Amazon
Keyspaces table and produces a JSON report plus Parquet snapshots of the
source set, the target set, and the set differences. It is read-only with
respect to both Cassandra and Amazon Keyspaces, does not touch
migration.ledger, the per-tile Iceberg snapshot tables, or the replication
DLQ, and is safe to run concurrently with an active replication job. This
release supports Keyspaces targets only.
The job is provisioned at cqlreplicator init time as a Glue job named
PrimaryKeyReconciliation${env}, using the same Glue version, Glue version,
IAM role, Glue connector, and artifacts/ JAR set as the replicator's job.
The reconciliation Glue job is created with MaxRetries = 0: a failed run
is not retried automatically by Glue; the operator reruns it explicitly.
The following invocation reconciles
ks_test_cql_replicator.test_cql_replicator on Cassandra against the same
keyspace/table on Amazon Keyspaces and prints the resulting Glue JobRunId:
cqlreplicator --cmd reconcile-keys \
--landing-zone s3://<PROVIDED_BUCKET_NAME> --region us-east-1 \
--src-keyspace ks_test_cql_replicator --src-table test_cql_replicator \
--trg-keyspace ks_test_cql_replicator --trg-table test_cql_replicator \
--target-type keyspaces --env ver5When the run finishes, inspect
s3://<PROVIDED_BUCKET_NAME>/ks_test_cql_replicator/test_cql_replicator/reconciliation/<Run_ID>/report.json.
The reconciliation target scan reads at LOCAL_ONE and honors --max-target-rcus
(the MAX_TARGET_RCUS Spark Cassandra Connector read-rate cap). For most runs
this is enough. If a few hot partitions still get throttled — you will see
ReadTimeoutException in the Glue logs, and the failure analyzer may report
Failed to open native connection to Cassandra — the DataStax driver's default
retry policy may not be conservative enough.
In that case you can install the AWS Keyspaces sample AmazonKeyspacesExponentialRetryPolicy, which retries on the same host with
exponential backoff (instead of the default RETRY_NEXT) and is well suited to
capacity-driven Keyspaces timeouts.
Important — this must be a JAR, not inlined in the script. The Spark Cassandra Connector opens Cassandra sessions on the Spark executors, and the driver loads
advanced.retry-policy.classby name on each executor. A class compiled into the Glue Scala script lives only on the driver, so referencing it fromKeyspacesConnector.conffails on executors withCan't find class com.aws.ssa.keyspaces.retry.AmazonKeyspacesExponentialRetryPolicy. The class must be on the executor classpath, which means it has to be a JAR supplied via--extra-jars.
The policy lives in the AWS Keyspaces driver-helpers samples. Build it (or pull
the prebuilt artifact) so you get a small JAR containing
com.aws.ssa.keyspaces.retry.AmazonKeyspacesExponentialRetryPolicy and
com.aws.ssa.keyspaces.retry.KeyspacesRetryOption:
git clone https://github.com/aws-samples/amazon-keyspaces-java-driver-helpers.git
cd amazon-keyspaces-java-driver-helpers
mvn -q -DskipTests package
# produces target/amazon-keyspaces-helpers-<version>.jarThe DataStax driver must be provided (it is already on the Glue classpath via
spark-cassandra-connector-assembly), so the helper JAR stays thin and does not
shade a conflicting driver.
Upload it next to the connector JARs the init flow publishes:
aws s3 cp target/amazon-keyspaces-helpers-<version>.jar \
s3://<PROVIDED_BUCKET_NAME>/artifacts/amazon-keyspaces-helpers-<version>.jar \
--region us-east-1The reconciliation job (PrimaryKeyReconciliation${env}) and the replicator job
(CQLReplicator${env}) share KeyspacesConnector.conf, so the JAR must be on the
classpath of both (driver and executors). Update each job's --extra-jars
default argument to include the new JAR (keep all existing entries and append the
helper JAR, comma separated):
aws glue update-job --job-name PrimaryKeyReconciliation${env} --region us-east-1 \
--job-update '{"DefaultArguments":{"--extra-jars":"<existing jars>,s3://<PROVIDED_BUCKET_NAME>/artifacts/amazon-keyspaces-helpers-<version>.jar", ...}}'Glue distributes --extra-jars to the driver and every executor, which is what
makes the class resolvable on the executor sessions the connector opens.
Add the advanced.retry-policy block and re-upload the conf to
s3://<PROVIDED_BUCKET_NAME>/artifacts/KeyspacesConnector.conf:
advanced.retry-policy {
class = com.aws.ssa.keyspaces.retry.AmazonKeyspacesExponentialRetryPolicy
max-attempts = 16
min-wait = 10 milliseconds
max-wait = 100 milliseconds
}On the next reconciliation run the driver logs the resolved policy:
target driver retry policy (advanced.retry-policy.class): com.aws.ssa.keyspaces.retry.AmazonKeyspacesExponentialRetryPolicy
If you instead see Can't find class ... on an executor, the JAR is not on
--extra-jars (step 3) — fix that before re-running. To roll back, remove the
advanced.retry-policy block from KeyspacesConnector.conf; the driver falls
back to its default retry policy and no JAR is required.
This tool licensed under the Apache-2 License. See the LICENSE file.