Migrate data from an ApsaraDB for MongoDB instance to an ApsaraMQ for Kafka instance

Source database

• The server must have enough outbound bandwidth. Low bandwidth reduces migration speed.

• Migrate up to 1,000 collections per task. Exceeding this limit causes a request error — split into multiple tasks instead.

• DTS cannot connect to MongoDB over an SRV endpoint.

• DTS cannot migrate data from the admin, config, or local databases.

• For incremental data migration, one of the following requirements must be met: If DTS cannot access logs within that window, the task may fail and data inconsistency or loss may occur. This is not covered by the DTS service level agreement (SLA).

  • The oplog feature is enabled for the source database and operation logs are retained for at least 7 days.

  • Change streams are enabled for the source database and DTS can use change streams to subscribe to data changes within the last 7 days.

• For an inelastic Amazon DocumentDB cluster: enable change streams, and set Migration Method to ChangeStream and Architecture to Sharded Cluster.

• If the source is a sharded cluster:

  • The _id field in all collections must be unique; otherwise, data inconsistency may occur.

  • The number of mongos nodes cannot exceed 10.

  • The source must not contain orphaned documents. See the MongoDB documentation and the DTS FAQ.

  • During full data migration, disable the balancer until each subtask reaches the incremental migration phase.

  • If Migration Method is Oplog, DTS cannot guarantee write order across different shards.

  • If the balancer has balancing data behavior, it may cause latency.

• During full data migration: do not make schema changes (including array type updates) and do not write to the source database. Both can cause data inconsistency.

Other limitations

• Only collections can be selected as migration objects — not databases.

• Single document size limit: 10 MB. Tasks fail if exceeded.

• Transactions are not retained; each transaction is converted to a single record in the destination.

• If broker nodes are added or removed in the destination Kafka instance during a running DTS task, restart the task.

• DTS must be able to connect to both the source and destination. Check that security settings, and the listeners and advertised.listeners parameters in the Kafka server.properties file, do not block DTS.

• Full data migration uses read and write resources on both databases. Run migration during off-peak hours (for example, when CPU utilization is below 30%).

• DTS attempts to recover failed instances for up to 7 days. Stop or release the task before switching workloads to the destination — otherwise the destination data may be overwritten if the task resumes automatically.

• Incremental migration latency is calculated based on the latest migrated data timestamp vs. the current source timestamp. If the source has no updates for a long period, the latency display may be inaccurate. Trigger an update operation on the source to refresh it.

• If a DTS task fails, DTS technical support tries to restore it within 8 hours. The task may be restarted and task parameters may be modified during restoration. Only task parameters are changed — database parameters are not.

Full data migration

DTS migrates existing data from all selected collections.

Incremental data migration

Step 1: Go to the Data Migration page

Use one of the following methods.

Step 2: Configure source and destination databases

Click Create Task, then configure the following parameters.

Source database

Destination database

Step 3: Test connectivity

Click Test Connectivity and Proceed.

DTS server CIDR blocks must be added (automatically or manually) to the security settings of your databases. See Add the CIDR blocks of DTS servers.

For self-managed databases not using Alibaba Cloud Instance as the access method, click Test Connectivity in the CIDR Blocks of DTS Servers dialog box.

Step 4: Configure objects to migrate

On the Configure Objects page, set the following parameters.

Parameter	Description
Migration Types	Select Full Data Migration only, or both Full Data Migration and Incremental Data Migration for continuous sync.
Processing Mode of Conflicting Tables	Precheck and Report Errors: fails the precheck if identically named collections exist in the destination. Use object name mapping to rename collections if needed. Ignore Errors and Proceed: skips the precheck. DTS does not migrate records whose primary keys match existing records in the destination. Warning selecting this option risks data inconsistency.
Data Format in Kafka	Canal JSON only.
Kafka Data Compression Format	LZ4 (default — low compression ratio, high speed), GZIP (high compression ratio, low speed — high CPU usage), or Snappy (balanced).
Policy for Shipping Data to Kafka Partitions	See Specify the policy for migrating data to Kafka partitions.
Message acknowledgement mechanism	See Message acknowledgement mechanism.
Capitalization of Object Names in Destination Instance	Default: DTS default policy. See Specify the capitalization of object names in the destination instance.
Source Objects	Select collections from the Source Objects section, then click to add them to Selected Objects.

Click Next: Advanced Settings.

Step 5: Configure advanced settings

Step 6: Save and run the precheck

Click Next: Save Task Settings and Precheck.

To preview OpenAPI parameters before saving, move the pointer over the button and click Preview OpenAPI parameters.

DTS runs a precheck before starting the migration. The task starts only after passing.

If any item fails, click View Details next to it, fix the issue, and click Precheck Again.

If an alert item can be safely ignored, click Confirm Alert Details > Ignore > OK, then click Precheck Again. Ignoring alerts may cause data inconsistency.

Step 7: Purchase and start the instance

1. Wait until Success Rate reaches 100%, then click Next: Purchase Instance.

2. On the Purchase Instance page, configure the following.

1. Read and agree to Data Transmission Service (Pay-as-you-go) Service Terms.

2. Click Buy and Start, then click OK.

View the task progress on the Data Migration page.

Full-only migration tasks stop automatically when complete (Completed status).

Tasks with incremental migration run continuously (Running status) and do not stop automatically.

Oplog (recommended)

{"data":[{"person":{"skills":["database","ai"],"name":"testName","age":18},"_id":{"$oid":"67d27da49591697476e1**"},"cid":"a"}],"database":"kafkadb","es":1741847972000,"gtid":null,"id":174184797200000**,"isDdl":false,"mysqlType":null,"old":null,"pkNames":["_id"],"serverId":null,"sql":null,"sqlType":null,"table":"kafka_test","ts":1741847973438,"type":"INSERT"}

{"data":[{"$set":{"person.age":20}}],"database":"kafkadb","es":1741848051000,"gtid":null,"id":174184805100000**,"isDdl":false,"mysqlType":null,"old":[{"_id":{"$oid":"67d27da49591697476e1**"}}],"pkNames":["_id"],"serverId":null,"sql":null,"sqlType":null,"table":"kafka_test","ts":1741848051984,"type":"UPDATE"}

{"data":[{"$set":{"salary":100.0}}],"database":"kafkadb","es":1741848146000,"gtid":null,"id":174184814600000**,"isDdl":false,"mysqlType":null,"old":[{"_id":{"$oid":"67d27da49591697476e1**"}}],"pkNames":["_id"],"serverId":null,"sql":null,"sqlType":null,"table":"kafka_test","ts":1741848147734,"type":"UPDATE"}

{"data":[{"$unset":{"salary":true}}],"database":"kafkadb","es":1741848207000,"gtid":null,"id":174184820700000**,"isDdl":false,"mysqlType":null,"old":[{"_id":{"$oid":"67d27da49591697476e1**"}}],"pkNames":["_id"],"serverId":null,"sql":null,"sqlType":null,"table":"kafka_test","ts":1741848208186,"type":"UPDATE"}

{"data":[{"_id":{"$oid":"67d27da49591697476e1**"}}],"database":"kafkadb","es":1741848289000,"gtid":null,"id":174184828900000**,"isDdl":false,"mysqlType":null,"old":null,"pkNames":["_id"],"serverId":null,"sql":null,"sqlType":null,"table":"kafka_test","ts":1741848289798,"type":"DELETE"}

{"data":null,"database":"kafkadb","es":1741847893000,"gtid":null,"id":1741847893000000005,"isDdl":true,"mysqlType":null,"old":null,"pkNames":null,"serverId":null,"sql":{"drop":"kafka_test"},"sqlType":null,"table":"kafka_test","ts":1741847893760,"type":"DDL"}

ChangeStream — updated fields only

{"data":[{"person":{"skills":["database","ai"],"name":"testName","age":18},"_id":{"$oid":"67d27da49591697476e1**"},"cid":"a"}],"database":"kafkadb","es":1741847972000,"gtid":null,"id":174184797200000**,"isDdl":false,"mysqlType":null,"old":null,"pkNames":["_id"],"serverId":null,"sql":null,"sqlType":null,"table":"kafka_test","ts":1741847973803,"type":"INSERT"}

{"data":[{"$set":{"person.age":20}}],"database":"kafkadb","es":1741848051000,"gtid":null,"id":174184805100000**,"isDdl":false,"mysqlType":null,"old":[{"_id":{"$oid":"67d27da49591697476e1**"}}],"pkNames":["_id"],"serverId":null,"sql":null,"sqlType":null,"table":"kafka_test","ts":1741848052912,"type":"UPDATE"}

{"data":[{"$set":{"salary":100.0}}],"database":"kafkadb","es":1741848146000,"gtid":null,"id":174184814600000**,"isDdl":false,"mysqlType":null,"old":[{"_id":{"$oid":"67d27da49591697476e1**"}}],"pkNames":["_id"],"serverId":null,"sql":null,"sqlType":null,"table":"kafka_test","ts":1741848148056,"type":"UPDATE"}

{"data":[{"$unset":{"salary":1}}],"database":"kafkadb","es":1741848207000,"gtid":null,"id":174184820700000**,"isDdl":false,"mysqlType":null,"old":[{"_id":{"$oid":"67d27da49591697476e1**"}}],"pkNames":["_id"],"serverId":null,"sql":null,"sqlType":null,"table":"kafka_test","ts":1741848209142,"type":"UPDATE"}

{"data":[{"_id":{"$oid":"67d27da49591697476e1**"}}],"database":"kafkadb","es":1741848289000,"gtid":null,"id":174184828900000**,"isDdl":false,"mysqlType":null,"old":null,"pkNames":["_id"],"serverId":null,"sql":null,"sqlType":null,"table":"kafka_test","ts":1741848290254,"type":"DELETE"}

{"data":null,"database":"kafkadb","es":1741847893000,"gtid":null,"id":174184789300000****,"isDdl":true,"mysqlType":null,"old":null,"pkNames":null,"serverId":null,"sql":{"drop":"kafka_test"},"sqlType":null,"table":"kafka_test","ts":1741847894679,"type":"DDL"}

ChangeStream — full document on update

{"data":[{"person":{"skills":["database","ai"],"name":"testName","age":18},"_id":{"$oid":"67d27da49591697476e1**"},"cid":"a"}],"database":"kafkadb","es":1741847972000,"gtid":null,"id":174184797200000**,"isDdl":false,"mysqlType":null,"old":null,"pkNames":["_id"],"serverId":null,"sql":null,"sqlType":null,"table":"kafka_test","ts":1741847973128,"type":"INSERT"}

{"data":[{"person":{"skills":["database","ai"],"name":"testName","age":20},"_id":{"$oid":"67d27da49591697476e1**"},"cid":"a"}],"database":"kafkadb","es":1741848051000,"gtid":null,"id":174184805100000,"isDdl":false,"mysqlType":null,"old":[{"_id":{"$oid":"67d27da49591697476e1**"}}],"pkNames":["_id"],"serverId":null,"sql":null,"sqlType":null,"table":"kafka_test","ts":1741848052219,"type":"UPDATE"}

{"data":[{"person":{"skills":["database","ai"],"name":"testName","age":20},"_id":{"$oid":"67d27da49591697476e1**"},"salary":100.0,"cid":"a"}],"database":"kafkadb","es":1741848146000,"gtid":null,"id":174184814600000,"isDdl":false,"mysqlType":null,"old":[{"_id":{"$oid":"67d27da49591697476e1**"}}],"pkNames":["_id"],"serverId":null,"sql":null,"sqlType":null,"table":"kafka_test","ts":1741848147327,"type":"UPDATE"}

{"data":[{"person":{"skills":["database","ai"],"name":"testName","age":20},"_id":{"$oid":"67d27da49591697476e1**"},"cid":"a"}],"database":"kafkadb","es":1741848207000,"gtid":null,"id":174184820700000,"isDdl":false,"mysqlType":null,"old":[{"_id":{"$oid":"67d27da49591697476e1**"}}],"pkNames":["_id"],"serverId":null,"sql":null,"sqlType":null,"table":"kafka_test","ts":1741848208401,"type":"UPDATE"}

{"data":[{"_id":{"$oid":"67d27da49591697476e1**"}}],"database":"kafkadb","es":1741848289000,"gtid":null,"id":174184828900000**,"isDdl":false,"mysqlType":null,"old":null,"pkNames":["_id"],"serverId":null,"sql":null,"sqlType":null,"table":"kafka_test","ts":1741848290499,"type":"DELETE"}

{"data":null,"database":"kafkadb","es":1741847893000,"gtid":null,"id":174184789300000****,"isDdl":true,"mysqlType":null,"old":null,"pkNames":null,"serverId":null,"sql":{"drop":"kafka_test"},"sqlType":null,"table":"kafka_test","ts":1741847894045,"type":"DDL"}

Special case: missing fullDocument

When Obtain the entire document after it is updated is Yes, if the fullDocument field is missing from an update event (for example, in a sharded collection), the delivery result is the same as the Oplog scenario — only the changed fields are sent.

Example

{"data":[{"$set":{"name":"b"}}],"database":"dts_test","es":1740720994000,"gtid":null,"id":174072099400000****,"isDdl":false,"mysqlType":null,"old":[{"name":"a","_id":1.0}],"pkNames":["_id"],"serverId":null,"sql":null,"sqlType":null,"table":"cstest","ts":1740721007099,"type":"UPDATE"}