Data Transmission Service:Synchronize data from an ApsaraDB for MongoDB sharded cluster instance to a Function Compute function

What your function receives

Each invocation passes an object with a Records array. Every element in the array represents one change event:

The function receives two categories of operations:

• DDL — schema changes: CreateIndex, CreateCollection, DropIndex, DropCollection

• DML — data changes: INSERT, UPDATE, DELETE

db.createCollection("testCollection")

{
  "Records": [{
    "data": [{"doc": "{\"create\": \"testCollection\", \"idIndex\": {\"v\": 2, \"key\": {\"_id\": 1}, \"name\": \"_id_\"}}"}],
    "pkNames": ["_id"],
    "type": "DDL",
    "es": 1694056437000,
    "database": "MongoDBTest",
    "id": 0,
    "isDdl": true,
    "table": "testCollection",
    "ts": 1694056437510
  }]
}

db.testCollection.drop()

{
  "Records": [{
    "data": [{"doc": "{\"drop\": \"testCollection\"}"}],
    "pkNames": ["_id"],
    "type": "DDL",
    "es": 1694056577000,
    "database": "MongoDBTest",
    "id": 0,
    "isDdl": true,
    "table": "testCollection",
    "ts": 1694056577789
  }]
}

db.testCollection.createIndex({name: 1})

{
  "Records": [{
    "data": [{"doc": "{\"createIndexes\": \"testCollection\", \"v\": 2, \"key\": {\"name\": 1}, \"name\": \"name_1\"}"}],
    "pkNames": ["_id"],
    "type": "DDL",
    "es": 1694056670000,
    "database": "MongoDBTest",
    "id": 0,
    "isDdl": true,
    "table": "testCollection",
    "ts": 1694056670719
  }]
}

db.testCollection.dropIndex({name: 1})

{
  "Records": [{
    "data": [{"doc": "{\"dropIndexes\": \"testCollection\", \"index\": \"name_1\"}"}],
    "pkNames": ["_id"],
    "type": "DDL",
    "es": 1694056817000,
    "database": "MongoDBTest",
    "id": 0,
    "isDdl": true,
    "table": "$cmd",
    "ts": 1694056818035
  }]
}

// Batch insert
db.runCommand({insert: "user", documents: [{"name": "jack", "age": 20}, {"name": "lili", "age": 20}]})

// Single insert
db.user.insert({"name": "jack", "age": 20})

{
  "Records": [
    {
      "data": [{"doc": "{\"_id\": {\"$oid\": \"64f9397f6e255f74d65a****\"}, \"name\": \"jack\", \"age\": 20}"}],
      "pkNames": ["_id"],
      "type": "INSERT",
      "es": 1694054783000,
      "database": "MongoDBTest",
      "id": 0,
      "isDdl": false,
      "table": "user",
      "ts": 1694054784427
    },
    {
      "data": [{"doc": "{\"_id\": {\"$oid\": \"64f9397f6e255f74d65a****\"}, \"name\": \"lili\", \"age\": 20}"}],
      "pkNames": ["_id"],
      "type": "INSERT",
      "es": 1694054783000,
      "database": "MongoDBTest",
      "id": 0,
      "isDdl": false,
      "table": "user",
      "ts": 1694054784428
    }
  ]
}

db.user.update({"name": "jack"}, {$set: {"age": 30}})

{
  "Records": [{
    "data": [{"doc": "{\"$set\": {\"age\": 30}}"}],
    "pkNames": ["_id"],
    "old": [{"doc": "{\"_id\": {\"$oid\": \"64f9397f6e255f74d65a****\"}}"}],
    "type": "UPDATE",
    "es": 1694054989000,
    "database": "MongoDBTest",
    "id": 0,
    "isDdl": false,
    "table": "user",
    "ts": 1694054990555
  }]
}

db.user.remove({"name": "jack"})

{
  "Records": [{
    "data": [{"doc": "{\"_id\": {\"$oid\": \"64f9397f6e255f74d65a****\"}}"}],
    "pkNames": ["_id"],
    "type": "DELETE",
    "es": 1694055452000,
    "database": "MongoDBTest",
    "id": 0,
    "isDdl": false,
    "table": "user",
    "ts": 1694055452852
  }]
}

Limitations

Review these constraints before creating your sync task.

Scope constraints:

• Only incremental data synchronization is supported. Full data synchronization is not supported.

• Cross-region synchronization is not supported.

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

• Object mapping is not supported.

• Transaction information is not retained. Transactions are converted to individual records at the destination.

Source database constraints:

• The source must use the sharded cluster architecture, with no more than 10 Mongos nodes.

• The source cannot be an Azure Cosmos DB for MongoDB cluster or an Amazon DocumentDB elastic cluster.

• The collections to be synchronized must have PRIMARY KEY or UNIQUE constraints and all fields must be unique. Otherwise, the destination database may contain duplicate data records.

• A single document cannot exceed 16 MB. Documents over this size cannot be written to the destination function and trigger an error. Use the extract, transform, and load (ETL) feature to filter large fields if needed.

• Synchronize up to 1,000 collections per task. For more collections, create multiple tasks or synchronize at the database level.

• The source instance cannot be scaled while a sync task is running.

• DTS cannot connect to a MongoDB database over an SRV endpoint.

• If the source database balancer is active, the task may experience delays.

• If the source database is a self-managed MongoDB database that uses the sharded cluster architecture, set the Access Method parameter to Express Connect, VPN Gateway, or Smart Access Gateway or Cloud Enterprise Network (CEN).

Write constraints:

• For INSERT operations, the data being inserted must contain shard keys.

• For UPDATE operations, shard keys cannot be modified.

• Configure only one DTS task per destination function. Multiple tasks writing to the same function may cause data errors.

Oplog and change stream requirements:

• The oplog must be enabled and retain at least 7 days of log data, OR change streams must be enabled to cover at least the last 7 days of changes. If neither condition is met, DTS may fail to capture source changes, which can cause data inconsistency or loss not covered by the DTS service level agreement (SLA).

Change stream restrictions (if applicable):

• Change streams require MongoDB 4.0 or later.

• Two-way synchronization is not supported when using change streams.

• For non-elastic Amazon DocumentDB clusters, use change streams: set Migration Method to ChangeStream and Architecture to Sharded Cluster.

New databases:

• DTS does not synchronize incremental data from databases created after the sync task starts.

Supported operations

The operations captured depend on the migration method.

Using oplog (recommended):

• CREATE COLLECTION, CREATE INDEX

• DROP DATABASE, DROP COLLECTION, DROP INDEX

• RENAME COLLECTION

• Document-level INSERT, UPDATE, and DELETE

Using change streams:

• DROP DATABASE, DROP COLLECTION

• RENAME COLLECTION

• Document-level INSERT, UPDATE, and DELETE

Billing

Incremental data synchronization is charged. For pricing details, see Billing overview.

DTS charges for the instance during connection retry periods.

Prerequisites

Before you begin, make sure that you have:

• A running ApsaraDB for MongoDB sharded cluster instance. See Create a sharded cluster instance

• A Function Compute service and function, with Handler Type set to Event Handler. See Quickly create a function

• A database account on the source ApsaraDB for MongoDB instance with read permissions on the source, admin, and local databases. See Manage the permissions of MongoDB database users

Create a sync task

Step 4: Select objects to synchronize

In the Configure Objects step, configure the following:

Parameter	Description
Synchronization Types	Fixed to Incremental Data Synchronization. Cannot be changed
Data Format	Fixed to Canal Json. For field descriptions, see the Canal Json section of the Data formats of a Kafka cluster topic
Source Objects	Select databases or collections to synchronize, then click to move them to Selected Objects
Selected Objects	Review the selected objects. Click to remove objects. Right-click an object to configure synchronization granularity (database or collection level)

What's next

• Write your Function Compute event handler to process the Canal JSON payload. Use the isDdl field to branch between DDL and DML logic, and deserialize the doc string in each data element to access the document fields.

• To reduce the load of large-field documents on your function, configure ETL filters before the data reaches the function. See Configure ETL in a data migration or data synchronization task.

• To monitor synchronization health, configure alerting in Advanced Settings or after the task is created. See Configure monitoring and alerting when you create a DTS task.

• If a task fails, DTS technical support will attempt to restore it within 8 hours. During restoration, the task may be restarted and task parameters (not database parameters) may be adjusted. For a list of parameters that may be modified, see Modify instance parameters.