ApsaraDB for MongoDB:Capturer les modifications de données en temps réel à l'aide des flux de changement MongoDB

Écouter des événements DDL supplémentaires

Prérequis

MongoDB 6.0 ou version ultérieure (Guide de mise à niveau).

Procédure

1. Utilisez mongosh pour vous connecter à la base de données.

2. Exécutez la commande watch avec l'option showExpandedEvents: true :

   // mongo shell or mongosh v1.x
   cursor = db.getSiblingDB("test").watch([],
     {
       showExpandedEvents: true       // Enable listening for more DDL events
     }
   );
   cursor.next();

Vérification des résultats

1. Dans une nouvelle fenêtre SQL, exécutez une instruction de modification, telle que db.createCollection("myCollection1").

2. Consultez la fenêtre SQL d'origine pour vérifier la sortie liée à l'instruction exécutée.

   mg xxx test> cursor = db.getSiblingDB("test").watch([],
   ... {
   ...     showExpandedEvents: true     // Enable more DDL event listening
   ... }
   ... );
   ... cursor.next();
   Warning: If there are no documents in the batch, next will block. Use tryNext if you want to check if there are any documents without waiting.
   {
     _id: {
       _data: '8268777614000000012B042C0100296E5A100434919C64814940F3A61F9D97A3F60D5C463C6Fxxx03C63726561746500466F70657274696F6E4465737372697074696F6E0046466964496E6465787800461E76002B04466B657990046E5F6964002B02003C6E616D65003C5F69645F000000004'
     },
     operationType: 'create',
     clusterTime: Timestamp({ t: 1752659476, i: 1 }),
     collectionUUID: UUID('34919c64-8149-40f3-a61f-9xxx'),
     wallTime: ISODate('2025-07-16T09:51:16.043Z'),
     ns: { db: 'test', coll: 'myCollection1' },
     operationDescription: { idIndex: { v: 2, key: { _id: 1 }, name: '_id_' } }
   }

Activation des pré-images

Une pré-image dans MongoDB correspond à un snapshot complet d'un document avant sa modification ou sa suppression. Elle enregistre les valeurs d'origine avant qu'une modification ne survienne.

Prérequis

MongoDB 6.0 ou version ultérieure (Guide de mise à niveau).

Procédure

1. Activez les pré-images au niveau de la base de données :

   db.adminCommand({
     setClusterParameter: {
       changeStreamOptions: {
         preAndPostImages: { expireAfterSeconds: "off" } // "off" uses oplog retention period
       }
     }
   })

2. Activez les pré-images au niveau de la collection :

   Remarque

   Pour activer les pré-images pour toutes les collections d'une base de données, activez d'abord cette option au niveau de la base de données. Ensuite, activez-la individuellement pour chaque collection de cette base de données.

   Modifier une collection existante

   db.runCommand({
     collMod: "myCollection",
     changeStreamPreAndPostImages: { enabled: true }
   })

   Spécifier lors de la création d'une nouvelle collection

   db.createCollection("myCollection", { changeStreamPreAndPostImages: { enabled: true }})

3. Créez un écouteur de flux de modifications (spécifiez les options de pré-image) :

   // Create a listener on the target collection
   cursor = db.getCollection("myCollection").watch([],
     {
       fullDocument: 'required', // or 'whenAvailable'
       fullDocumentBeforeChange: 'required' // or 'whenAvailable'
     }
   )
   cursor.next();

   • required : le serveur doit renvoyer la pré/post-image. Sinon, il renvoie une erreur.

   • whenAvailable : le serveur tente de renvoyer l'image, mais sans garantie.

Vérification des résultats

1. Vérifiez si le paramètre au niveau de la base de données est activé :

   db.adminCommand( { getClusterParameter: "changeStreamOptions" } )

   En cas de succès, la commande renvoie une sortie similaire à la suivante :

   test> db.adminCommand( { getClusterParameter: "changeStreamOptions" } )
   {
     clusterParameters: [
       {
         _id: 'changeStreamOptions',
         clusterParameterTime: Timestamp({ t: 1752655937, i: 1 }),
         preAndPostImages: { expireAfterSeconds: Long('100') }
       }
     ],
     ok: 1,
     '$clusterTime': {
       clusterTime: Timestamp({ t: 1752656717, i: 1 }),
       signature: {
         hash: Binary.createFromBase64('xxx=', 0),
         keyId: Long('xxx')
       }
     },
     operationTime: Timestamp({ t: 1752656717, i: 1 })
   }

2. Vérifiez la configuration de la collection :

   db.getCollectionInfos({name: "myCollection"}) // or db.runCommand({listCollections: 1})

   Sortie attendue : recherchez un champ similaire à "options" : { "changeStreamPreAndPostImages" : { "enabled" : true } } dans le document renvoyé.

   test> db.getCollectionInfos({name: "myCollection"})
   [
     {
       name: 'myCollection',
       type: 'collection',
       options: { changeStreamPreAndPostImages: { enabled: true } },
       info: {
         readOnly: false,
         uuid: UUID('55ed1b7c-7575-4ba3-8afa-xxx')
       },
       idIndex: { v: 2, key: { _id: 1 }, name: '_id_' }
     }
   ]

3. Dans une autre fenêtre mongosh, mettez à jour un document dans myCollection.

4. Observez l'événement renvoyé par cursor. Il doit inclure le champ fullDocumentBeforeChange (le document avant la modification).

   ... cursor.next();
   {
     _id: {
       _data: '8268786B15000000012B042C0100296E5A100455ED1B7C75754BA38AFA0C5B56A360BC463C6F7065726174696F6E54797065003C7570646174650046646F63756D656E744B65790046xxx697CDDF6EE279xxx'
     },
     operationType: 'update',
     clusterTime: Timestamp({ t: 1752722197, i: 1 }),
     wallTime: ISODate('2025-07-17T03:16:37.626Z'),
     fullDocument: {
       _id: ObjectId('6878697cddfxxx'),
       name: 'test',
       count: 111
     },
     ns: { db: 'test', coll: 'myCollection' },
     documentKey: { _id: ObjectId('6878697cddfxxx') },
     updateDescription: {
       updatedFields: { count: 111 },
       removedFields: [],
       truncatedArrays: []
     },
     fullDocumentBeforeChange: { _id: ObjectId('6878697cddf6xxx'), name: 'test' }
   }

Pour plus d'informations, consultez la documentation Change Streams with Document Pre- and Post-Images.

Activer les post-images

Une post-image dans MongoDB correspond à un snapshot complet d'un document après qu'une modification a eu lieu. Elle enregistre l'intégralité du contenu du document tel qu'il est après la modification.

Prérequis

MongoDB 3.6 ou version ultérieure (Guide de mise à niveau).

Procédure

Lors de l'exécution de la commande watch, définissez fullDocument: 'updateLookup'.

cursor = db.getSiblingDB("test").myCollection.watch([], 
  {
    fullDocument: 'updateLookup'
  }
);
cursor.next();
              

Vérifiez les résultats

1. Dans une autre fenêtre mongosh, insérez ou mettez à jour un document dans myCollection.

2. Observez l'événement renvoyé par cursor. Il doit inclure le champ fullDocument (le document après la modification).

   xxx [primary] test> cursor = db.getSiblingDB("test").myCollection.watch([],
   ... {
   ...     fullDocument: 'updateLookup'
   ... }
   ... );
   ... cursor.next();
   [...
   {
     _id: {
       _data: 'xxx100296E5A10040D6C3FBC28484F08240555576066945463C6F7065726174696F6E54797065003C757064617465500046646F63756D656E744B6579005F6964004B657390046645F696400646881ADE741E7D4B638ED7CA1000004'
     },
     operationType: 'update',
     clusterTime: Timestamp({ t: 1753329166, i: 1 }),
     wallTime: ISODate('2025-07-24T03:52:46.966Z'),
     fullDocument: {
       _id: ObjectId('xxx8ed7ca1'),
       name: 'test1',
       age: 12,
       count: 2222
     },
     ns: { db: 'test', coll: 'myCollection' },
     documentKey: { _id: ObjectId('xxx7ca1') },
     updateDescription: {
       updatedFields: { count: 2222 },
       removedFields: [],
       truncatedArrays: []
     }
   }]

Gérer les événements de changement très volumineux (>16 Mo)

Prérequis

MongoDB 7.0 ou version ultérieure (Guide de mise à niveau).

Procédure

Incluez l'étape $changeStreamSplitLargeEvent dans le pipeline de la commande watch() :

myChangeStreamCursor = db.myCollection.watch(
  [ { $changeStreamSplitLargeEvent: {} } ], // Add split stage
  {
    fullDocument: "required",
    fullDocumentBeforeChange: "required"
  }
)

Vérifiez les résultats

• Exécutez une opération qui génère un événement de changement supérieur à 16 Mo (par exemple, la mise à jour d'un document contenant un tableau très volumineux).

• Notez que le flux d'événements renvoyé est divisé en plusieurs événements fragment consécutifs, se terminant par un dernier événement fragment.

Réduire la surcharge de stockage des pré-images

Par défaut, les pré-images expirent en même temps que l'oplog. Définissez une durée d'expiration plus courte pour économiser de l'espace :

db.adminCommand({
  setClusterParameter: {
    changeStreamOptions: {
      preAndPostImages: { expireAfterSeconds: 100 } // Unit: seconds
    }
  }
})

**1. Pourquoi les journaux de requêtes lentes des flux de modifications affichent-ils toujours COLLSCAN ?**

Ce comportement est normal et attendu. Aucune optimisation n'est nécessaire. Un curseur de flux de modifications lit en définitive depuis local.oplog.rs (le seul emplacement qui enregistre toutes les modifications de l'instance). Cette collection ne possède aucun index, elle effectue donc systématiquement une opération COLLSCAN. Il est impossible d'éviter ce mécanisme et aucune optimisation n'est possible à ce niveau.

Si vous utilisez le mot-clé COLLSCAN pour filtrer les journaux de requêtes lentes afin d'identifier les requêtes à optimiser, veillez également à exclure le mot-clé $changeStream.

N'examinez les journaux de requêtes lentes liés aux flux de modifications que si vous rencontrez des problèmes de performances (par exemple, une latence croissante lors de la consommation en aval).

2. Pourquoi, sur une instance sharding, vois-je des curseurs de flux de modifications sur des shards autres que le shard principal lorsque j'écoute une collection non partitionnée ?

Cette mesure prépare d'éventuelles opérations shardCollection. Vous pouvez convertir à tout moment une collection non partitionnée disposant d'un écouteur actif en une collection partitionnée, en répartissant ses données sur tous les shards (de manière similaire à movePrimary). La création anticipée de curseurs de flux de modifications sur les autres shards permet de gérer ce scénario. En règle générale, ces curseurs situés sur les autres shards ne renvoient aucun événement de modification et n'engendrent qu'une charge de performance minimale.

De même, afin de gérer les potentielles opérations addShard/removeShard, lorsque vous configurez un écouteur sur une instance sharding, mongos crée également les curseurs correspondants sur le Config Server.

3. Pourquoi le journal de requête lente d'un curseur de flux de modifications apparaît-il sur le nœud principal alors que j'ai spécifié readPreference:secondary lors de la création du curseur ?

Un curseur reste « épinglé » à un nœud spécifique après sa création et ne migre pas automatiquement lorsque les rôles des nœuds changent. Une fois créé, un curseur de flux de modifications reste fixé sur un nœud mongod spécifique (principal ou secondaire) et continue de consommer les données via getMore. À la suite d'événements tels qu'un basculement primaire/secondaire, un redimensionnement d'instance ou une migration, un curseur initialement situé sur un secondaire peut se retrouver sur le nœud principal.

Si vous souhaitez éviter que cette charge ne persiste sur le nœud principal, nettoyez les curseurs de flux de modifications concernés à l'aide de la commande killCursors. Votre logique de consommation en aval peut ensuite restaurer le curseur sur un nœud secondaire en utilisant le resumeToken et le paramètre readPreference. Cette opération ne provoque ni perte d'événements ni interruption.

db.runCommand(
   {
     killCursors: <collection>,
     cursors: [ <cursor id1>, ... ], comment: <any>
   }
)
db.getSiblingDB("<testDB>").runCommand( { killCursors: "<testColl>", cursors: [NumberLong("2452840976689696187") ] } ) 

4. Pourquoi observe-je de nombreux journaux de requêtes lentes concernant les flux de modifications sur mongos, durant 1000 ms ?

2020-08-26T04:34:45.045+0000 I COMMAND [conn21283] command altconfig-b2b-perf.oplog command: getMore \{ getMore: 3513599116181216748, collection: "oplog", $db: "altconfig-b2b-perf", $clusterTime: { clusterTime: Timestamp(1598416483, 1), signature: { hash: BinData(0, EC3841EB1FB7A34F897688BB5983E32E2ADF6763), keyId: 6855385090000683010 } }, lsid: \{ id: UUID("72864d11-c0f8-48bb-823e-de5f18a9c409") } } originatingCommand: \{ find: "oplog", filter: { timestamp: { $gte: new Date(1597895390009) } }, tailable: true, awaitData: true, $db: "altconfig-b2b-perf", $clusterTime: \{ clusterTime: Timestamp(1597895487, 1), signature: { hash: BinData(0, AAAE33D5688F935C70469CBDB8EB1F6882749A40), keyId: 6855385090000683010 } }, lsid: \{ id: UUID("72864d11-c0f8-48bb-823e-de5f18a9c409") } } nShards:1 cursorid:3513599116181216748 numYields:0 nreturned:0 reslen:237 protocol:op_msg 1000ms

Il s'agit d'un comportement normal dans les versions de MongoDB antérieures à la version 6.0. Cela indique un délai d'attente et non un goulot d'étranglement au niveau des performances. Dans les architectures de clusters sharding, mongos crée des curseurs de flux de modifications sur les shards avec les paramètres tailable:true, awaitData:true et maxTimeMS:1000 afin de renvoyer autant d'événements de modification que possible dans un délai de 1000 ms.

Dans les versions majeures antérieures à la 6.0, mongos enregistre ces journaux de requêtes lentes de 1000 ms, ce qui peut induire les utilisateurs en erreur. MongoDB a optimisé ce comportement. Pour plus de détails, consultez la page SERVER-50559.

5. Où sont stockées les données de pré-image après l'activation des pré-images ?

Les pré-images sont stockées dans la collection système config.system.preimages sur chaque nœud de données mongod (et non sur le Config Server) :

• Replica set : Stockées sur tous les nœuds de données (principal et secondaires). Les pré-images sont répliquées conjointement avec l'oplog.

• Cluster sharding : Stockées sur les nœuds mongod de chaque shard. Chaque shard stocke indépendamment les pré-images des documents qu'il traite.

Les post-images ne sont pas stockées séparément. Elles sont construites en temps réel à partir des données des événements de flux de modifications.

6. Quelle surcharge l'activation des pré-images entraîne-t-elle ?

• Surcharge de stockage : La collection config.system.preimages augmente continuellement. Sans politique d'expiration appropriée ou si la fenêtre de l'oplog est importante, elle peut consommer un espace disque significatif.

• Surcharge en écriture : Chaque opération update/replace/delete écrit une copie supplémentaire du document original complet dans config.system.preimages, doublant ainsi les E/S d'écriture (particulièrement notable pour les documents volumineux).

• Pression sur le cache WiredTiger : Les données présentes dans config.system.preimages entrent également dans le cache WiredTiger, entrant en concurrence avec les données applicatives pour l'espace cache disponible.

• Entrées d'oplog : Les écritures de pré-images génèrent des entrées d'oplog car elles doivent être répliquées vers les nœuds secondaires.

• Limite de 16 Mo : Pour les documents volumineux, les événements de changement incluant des pré-images peuvent dépasser la limite de taille BSON de 16 Mo, provoquant une erreur du flux de modifications. Dans MongoDB 6.0, l'étape $changeStreamSplitLargeEvent n'est pas encore disponible ; la base de données ne peut donc pas gérer cela automatiquement et nécessite une intervention au niveau de l'application.