Data Transmission Service:Synchronize data from ApsaraDB for MongoDB to Message Queue for Apache Kafka

Source database limits

• The source server must have enough outbound bandwidth. Insufficient bandwidth reduces synchronization speed.

• When configuring name mapping for collections, a single task can synchronize up to 1,000 collections. Exceeding this limit causes a request error. To synchronize more than 1,000 collections, create multiple tasks or synchronize the entire database without name mapping.

• For sharded cluster sources: the _id field in each synchronized collection must be unique, or data inconsistency may occur.

• For sharded cluster sources: the number of mongos nodes cannot exceed 10, and the instance must not contain orphaned documents. See the FAQ topic for how to remove orphaned documents.

• Standalone ApsaraDB for MongoDB instances, Azure Cosmos DB for MongoDB clusters, and Amazon DocumentDB elastic clusters are not supported as sources.

• DTS cannot connect to MongoDB over an SRV endpoint.

• The oplog must be enabled and retain log data for at least 7 days, OR change streams must be enabled so that DTS can subscribe to data changes from the last 7 days. If neither condition is met, DTS may fail to fetch source changes, and data loss or inconsistency may occur outside the scope of the service level agreement (SLA).

  Important

  - Use the oplog to record source database changes (recommended). - Change streams require MongoDB 4.0 or later. Two-way synchronization is not supported when using change streams. - For non-elastic Amazon DocumentDB clusters, enable change streams and set Migration Method to ChangeStream and Architecture to Sharded Cluster.

• During full data synchronization, do not modify database or collection schemas, or change data of the ARRAY type.

• For sharded cluster sources, do not run the following commands during synchronization: shardCollection, reshardCollection, unshardCollection, moveCollection, or movePrimary. These commands can cause data inconsistency.

• If the source database is a MongoDB instance that uses the sharded cluster architecture and the balancer of the source database balances data, latency may occur in the instance.

Other limits

• Only collection-level synchronization is supported.

• The admin, config, and local databases cannot be synchronized.

• A single record exceeding 10 MB causes the task to fail.

• Transactions are not preserved. DTS converts each transaction into individual records at the destination.

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

• Make sure that DTS can connect to the source and destination instances. For example, the security settings of the database instances and the listeners and advertised.listeners parameters in the server.properties file of a self-managed Kafka instance do not restrict access from DTS.

• Run synchronization during off-peak hours when CPU utilization on both source and destination databases is below 30%.

• DTS automatically retries failed instances that have been running for less than 7 days. Before switching traffic to the destination, stop or release the synchronization instance to prevent automatic resumption from overwriting destination data.

• DTS calculates the latency of incremental data synchronization based on the timestamp of the latest synchronized data in the destination database and the current timestamp in the source database. If no update operation is performed on the source database for an extended period of time, the synchronization latency may be inaccurate. If the latency of the data synchronization task is excessively high, you can perform an update operation on the source database to update the latency.

• If a DTS task fails to run, DTS technical support will try to restore the task within 8 hours. During the restoration, the task may be restarted, and the parameters of the task may be modified. Only the parameters of the DTS task may be modified. The parameters of databases are not modified. The parameters that may be modified include but are not limited to the parameters in the "Modify instance parameters" section.

• For sharded cluster sources with Oplog as the incremental sync method, DTS does not guarantee write order across shards to the target Kafka topic.

• For sharded cluster sources, disable the MongoDB balancer during full data synchronization. Re-enable it only after full sync completes and incremental sync begins. See Manage the ApsaraDB for MongoDB balancer.

Step 1: Open the Data Synchronization page

Use either the DTS console or the DMS console.

DTS console

1. Log on to the DTS console.DTS console

2. In the left-side navigation pane, click Data Synchronization.

3. In the upper-left corner, select the region where the synchronization task resides.

DMS console

Steps may vary depending on the DMS console mode. See Simple mode and Customize the layout and style of the DMS console.

1. Log on to the DMS console.DMS console

2. In the top navigation bar, move the pointer over Data + AI and choose DTS (DTS) > Data Synchronization.

3. From the drop-down list to the right of Data Synchronization Tasks, select the region.

Step 2: Configure source and destination databases

Click Create Task, then configure the parameters described in the following tables.

General

Source database

Destination database

Step 3: Test connectivity

Click Test Connectivity and Proceed at the bottom of the page.

DTS adds its server CIDR blocks to the security settings of the source and destination databases automatically, if allowed. For manual setup, 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 synchronize

In the Configure Objects step, set the following parameters.

Step 5: Configure advanced settings

Click Next: Advanced Settings and configure the following parameters.

Step 6: Run the precheck

Click Next: Save Task Settings and Precheck.

To preview the API parameters for this configuration, hover over the button and click Preview OpenAPI parameters before proceeding.

The task cannot start until the precheck passes.

If the precheck fails, click View Details next to each failed item, resolve the issue, then click Precheck Again.

If an alert is raised for an item: fix non-ignorable alerts before proceeding. For ignorable alerts, click Confirm Alert Details, then Ignore in the dialog, confirm, and click Precheck Again. Ignoring alerts may cause data inconsistency.

Step 7: Purchase and start the instance

1. Wait for Success Rate to reach 100%, then click Next: Purchase Instance.

2. On the buy page, configure the following parameters.

   Parameter	Description
   Billing Method	Subscription: pay upfront; more cost-effective for long-term use. Pay-as-you-go: billed hourly; suitable for short-term use. Release the instance when no longer needed to stop billing.
   Resource Group Settings	The resource group for the synchronization instance. Default: default resource group. See What is Resource Management?
   Instance Class	The synchronization speed tier. See Instance classes of data synchronization instances.
   Subscription Duration	Available for Subscription billing. Options: 1–9 months, 1 year, 2 years, 3 years, or 5 years.

3. Read and accept Data Transmission Service (Pay-as-you-go) Service Terms.

4. Click Buy and Start, then click OK in the confirmation dialog.

The task appears in the task list. Monitor its progress there.

Choose your scenario

Canal JSON message fields

All three scenarios use the same top-level Canal JSON envelope. The following fields appear in every message.

Scenario 1: Oplog

Set Migration Method to Oplog.

{
    "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"
}

Scenario 2: ChangeStream — updated fields only

Set Migration Method to ChangeStream. Set Obtain the entire document after it is updated to No.

Insert and delete messages are identical to Scenario 1. Update messages contain only the changed fields.

{
    "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": [{
        "$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"
}

Scenario 3: ChangeStream — full document on update

Set Migration Method to ChangeStream. Set Obtain the entire document after it is updated to Yes.

Update events deliver the full document after the change instead of just the changed fields.

{
    "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"
}

Special case: missing fullDocument in ChangeStream

When a ChangeStream update event's fullDocument field is missing — for example, when a document moves across shards in a sharded collection — the delivered message falls back to Oplog behavior (partial update with $set or $unset operators).

Example: sharded collection update where fullDocument is missing

Source base data:

use admin
db.runCommand({ enablesharding:"dts_test" })

Source incremental change:

use dts_test
sh.shardCollection("dts_test.cstest",{"name":"hashed"})
db.cstest.insert({"_id":1,"name":"a"})
db.cstest.updateOne({"_id":1,"name":"a"},{$set:{"name":"b"}})

{
    "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"
}