ApsaraDB for MongoDB:Use MongoDB change streams to capture data changes in real time

Listen for more DDL events

Prerequisites

MongoDB 6.0 or later (Upgrade guide).

Procedure

1. Connect to the database using mongosh.

2. Run the watch command and set showExpandedEvents: true:

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

Verify the result

1. In a new mongosh window, run a change command, such as db.createCollection("myCollection1").

2. You can view the SQL execution details in the Outputs of the original SQL window.

   

Enable pre-images

A MongoDB pre-image is a complete snapshot of a document before it is modified or deleted. It records the original value of the data before the change.

Prerequisites

MongoDB 6.0 or later (Upgrade guide).

Procedure

1. Enable pre-images at the database level:

   db.adminCommand({
     setClusterParameter: {
       changeStreamOptions: {
         preAndPostImages: { expireAfterSeconds: "off" } // "off" indicates that the oplog retention period is used
       }
     }
   })

2. Enable pre-images at the collection level:

   Note

   To enable pre-images for all collections in a database, you must first enable them at the database level. Then, you must enable the feature for each collection individually.

   Modify an existing collection

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

   Specify when creating a new collection

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

3. Create a change stream listener and specify the pre-image option:

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

   • required: The server must return a pre-image or post-image. Otherwise, an error is reported.

   • whenAvailable: The server attempts to return the image, but it is not guaranteed.

Verify the result

1. Check whether pre-images are enabled at the database level:

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

   If enabled, the following output is returned:

2. Check the collection configuration:

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

   Expected output: In the returned document, look for a field similar to "options" : { "changeStreamPreAndPostImages" : { "enabled" : true } }.

   

3. In another mongosh window, update a document in myCollection.

4. Observe the event returned by the cursor. It should contain the fullDocumentBeforeChange field, which is the pre-image.

   

For more information, see Change Streams with Document Pre- and Post-Images.

Enable post-images

A MongoDB post-image is a complete snapshot of a document after a change occurs. It records the full content of the document after the change.

Prerequisites

MongoDB 3.6 or later (Upgrade guide).

Procedure

When you run the watch command, set fullDocument: 'updateLookup'.

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

Verify the result

1. In another mongosh window, add or update a document in myCollection.

2. Observe the event returned by the cursor. It should contain the fullDocument field, which is the post-image.

Handle large change events (> 16 MB)

Prerequisites

MongoDB 7.0 or later (Upgrade guide).

Procedure

You can include the $changeStreamSplitLargeEvent stage in the watch() pipeline:

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

Verify the result

• Run an operation that causes a change event larger than 16 MB, such as updating a document that contains a very large array.

• Observe the returned event stream. It is split into multiple consecutive fragment events, ending with a final fragment event.

Reduce pre-image storage overhead

By default, pre-images expire with the oplog. You can set a shorter expiration time to save storage space:

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