Data Transmission Service:Synchroniser les données d'une instance ApsaraDB for MongoDB en cluster fragmenté vers une fonction Function Compute

Contenu reçu par votre fonction

Chaque invocation transmet un objet contenant un tableau Records. Chaque élément du tableau représente un événement de modification :

La fonction reçoit deux catégories d'opérations :

• DDL — modifications de schéma : CreateIndex, CreateCollection, DropIndex, DropCollection

• DML — modifications de données : 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

Vérifiez ces contraintes avant de créer votre tâche de synchronisation.

Contraintes de périmètre :

• Seule la synchronisation des données incrémentielles est prise en charge. La synchronisation complète des données n'est pas disponible.

• La synchronisation interrégionale n'est pas prise en charge.

• DTS ne peut pas synchroniser les données des bases de données admin, config ou local.

• Le mappage d'objets n'est pas pris en charge.

• Les informations de transaction ne sont pas conservées. Les transactions sont converties en enregistrements individuels dans la destination.

Contraintes liées à la base de données source :

• La source doit utiliser une architecture de cluster fragmenté, avec un maximum de 10 nœuds Mongos.

• La source ne peut pas être un cluster Azure Cosmos DB for MongoDB ni un cluster élastique Amazon DocumentDB.

• Les collections à synchroniser doivent disposer de contraintes PRIMARY KEY ou UNIQUE et tous les champs doivent être uniques. Sinon, la base de données de destination risque de contenir des enregistrements de données en double.

• Un document unique ne doit pas dépasser 16 Mo. Les documents dépassant cette taille ne peuvent pas être écrits dans la fonction de destination et déclenchent une erreur. Utilisez la fonctionnalité ETL (Extract, Transform, Load) pour filtrer les champs volumineux si nécessaire.

• Synchronisez jusqu'à 1 000 collections par tâche. Pour un nombre supérieur de collections, créez plusieurs tâches ou effectuez la synchronisation au niveau de la base de données.

• L'instance source ne peut pas être mise à l'échelle pendant l'exécution d'une tâche de synchronisation.

• DTS ne peut pas se connecter à une base de données MongoDB via un endpoint SRV.

• Si l'équilibreur de charge de la base de données source est actif, la tâche peut subir des retards.

• Si la base de données source est une base de données MongoDB auto-gérée utilisant une architecture de cluster fragmenté, définissez le paramètre Access Method sur Express Connect, VPN Gateway, or Smart Access Gateway ou Cloud Enterprise Network (CEN).

Contraintes d'écriture :

• Pour les opérations INSERT, les données insérées doivent contenir des clés de fragmentation (shard keys).

• Pour les opérations UPDATE, les clés de fragmentation ne peuvent pas être modifiées.

• Configurez une seule tâche DTS par fonction de destination. Plusieurs tâches écrivant dans la même fonction peuvent provoquer des erreurs de données.

Exigences relatives à l'oplog et aux flux de modifications (change streams) :

• L'oplog doit être activé et conserver au moins 7 jours de données de journal, OU les flux de modifications doivent être activés pour couvrir au moins les 7 derniers jours de modifications. Si aucune de ces conditions n'est remplie, DTS peut échouer à capturer les modifications sources, ce qui peut entraîner une incohérence ou une perte de données non couverte par l'accord de niveau de service (SLA) de DTS.

Restrictions liées aux flux de modifications (le cas échéant) :

• Les flux de modifications nécessitent MongoDB 4.0 ou une version ultérieure.

• La synchronisation bidirectionnelle n'est pas prise en charge lors de l'utilisation des flux de modifications.

• Pour les clusters Amazon DocumentDB non élastiques, utilisez les flux de modifications : définissez Migration Method sur ChangeStream et Architecture sur Sharded Cluster.

Nouvelles bases de données :

• DTS ne synchronise pas les données incrémentielles des bases de données créées après le démarrage de la tâche de synchronisation.

Opérations prises en charge

Les opérations capturées dépendent de la méthode de migration.

Utilisation de l'oplog (recommandé) :

• CREATE COLLECTION, CREATE INDEX

• DROP DATABASE, DROP COLLECTION, DROP INDEX

• RENAME COLLECTION

• Opérations INSERT, UPDATE et DELETE au niveau des documents

Utilisation des flux de modifications :

• DROP DATABASE, DROP COLLECTION

• RENAME COLLECTION

• Opérations INSERT, UPDATE et DELETE au niveau des documents

Facturation

La synchronisation des données incrémentielles est facturée. Pour plus de détails sur les tarifs, consultez Présentation de la facturation.

DTS facture l'instance pendant les périodes de nouvelle tentative de connexion.

Prérequis

Avant de commencer, assurez-vous d'avoir :

• Une instance ApsaraDB for MongoDB en cluster fragmenté en cours d'exécution. Consultez Créer une instance de cluster fragmenté

• Un service et une fonction Function Compute, avec Handler Type défini sur Event Handler. Consultez Créer rapidement une fonction

• Un compte de base de données sur l'instance ApsaraDB for MongoDB source disposant d'autorisations de lecture sur les bases de données source, admin et local. Consultez Gérer les autorisations des utilisateurs de base de données MongoDB

  Remarque

  Si vous utilisez ChangeStream comme méthode de synchronisation incrémentielle, le compte de la base de données source nécessite des autorisations de lecture Change Streams à l'échelle de l'instance (telles que readAnyDatabase). Si la source est une instance ApsaraDB for MongoDB avec un compte personnalisé, vous devez également accorder au compte l'autorisation de lecture sur la base de données admin. Pour plus de détails, consultez Autorisations du compte racine spécifié lors de la création de l'instance.

Créer une tâche de synchronisation

Étape 4 : Sélectionner les objets à synchroniser

À l'étape Configure Objects, configurez les éléments suivants :

Paramètre	Description
Synchronization Types	Fixé sur Incremental Data Synchronization. Ne peut pas être modifié
Data Format	Fixé sur Canal Json. Pour les descriptions des champs, consultez la section Canal Json des formats de données d'un sujet de cluster Kafka
Source Objects	Sélectionnez les bases de données ou les collections à synchroniser, puis cliquez sur pour les déplacer vers Selected Objects
Selected Objects	Vérifiez les objets sélectionnés. Cliquez sur pour supprimer des objets. Faites un clic droit sur un objet pour configurer la granularité de la synchronisation (niveau base de données ou collection)

Étapes suivantes

• Rédigez votre gestionnaire d'événements Function Compute pour traiter la charge utile Canal JSON. Utilisez le champ isDdl pour distinguer la logique DDL et DML, et désérialisez la chaîne doc dans chaque élément data pour accéder aux champs du document.

• Pour réduire la charge des documents contenant des champs volumineux sur votre fonction, configurez des filtres ETL avant que les données n'atteignent la fonction. Consultez Configurer l'ETL dans une tâche de migration ou de synchronisation de données.

• Pour surveiller l'état de santé de la synchronisation, configurez les alertes dans Advanced Settings ou après la création de la tâche. Consultez Configurer la surveillance et les alertes lors de la création d'une tâche DTS.

• En cas d'échec d'une tâche, l'assistance technique DTS tentera de la restaurer dans un délai de 8 heures. Pendant la restauration, la tâche peut être redémarrée et les paramètres de la tâche (et non ceux de la base de données) peuvent être ajustés. Pour obtenir la liste des paramètres susceptibles d'être modifiés, consultez Modifier les paramètres de l'instance.