Monitor Data with Change Streams

Overview

In this guide, you can learn how to use a change stream to monitor real-time changes to your database. A change stream is a MongoDB Server feature that allows your application to subscribe to data changes on a collection, database, or deployment.

Open a Change Stream

To open a change stream, call the watch() method. The instance on which you call the watch() method on determines the scope of events that the change stream listens for. You can call the watch() method on the following classes:

• MongoClient: To monitor all changes in the MongoDB deployment

• Database: To monitor changes in all collections in the database

• Collection: To monitor changes in the collection

The following example opens a change stream on the restaurants collection and outputs changes as they occur. Select the Synchronous or Asynchronous tab to see the corresponding code:

database = client["sample_restaurants"]
collection = database["restaurants"]
with collection.watch() as stream:
 for change in stream:
 print(change)

database = client["sample_restaurants"]
collection = database["restaurants"]
async with await collection.watch() as stream:
 async for change in stream:
 print(change)

To begin watching for changes, run the application. Then, in a separate application or shell, modify the restaurants collection. The following example updates a document with a name field value of Blarney Castle. Select the Synchronous or Asynchronous tab to see the corresponding code:

database = client["sample_restaurants"]
collection = database["restaurants"]
query_filter = { "name": "Blarney Castle" }
update_operation = { '$set' : 
 { "cuisine": "Irish" }
}
result = collection.update_one(query_filter, update_operation)

database = client["sample_restaurants"]
collection = database["restaurants"]
query_filter = { "name": "Blarney Castle" }
update_operation = { '$set' : 
 { "cuisine": "Irish" }
}
result = await collection.update_one(query_filter, update_operation)

When you update the collection, the change stream application prints the change as it occurs. The printed change event resembles the following:

{'_id': {'_data': '...'}, 'operationType': 'update', 'clusterTime': Timestamp(...), 'wallTime': datetime.datetime(...),
'ns': {'db': 'sample_restaurants', 'coll': 'restaurants'}, 'documentKey': {'_id': ObjectId('...')},
'updateDescription': {'updatedFields': {'cuisine': 'Irish'}, 'removedFields': [], 'truncatedArrays': []}}

Modify the Change Stream Output

You can pass the pipeline parameter to the watch() method to modify the change stream output. This parameter allows you to watch for only specified change events. Format the parameter as a list of objects that each represent an aggregation stage.

You can specify the following stages in the pipeline parameter:

• $addFields

• $match

• $project

• $replaceRoot

• $replaceWith

• $redact

• $set

• $unset

The following example uses the pipeline parameter to open a change stream that records only update operations. Select the Synchronous or Asynchronous tab to see the corresponding code:

change_pipeline = { "$match": { "operationType": "update" }},
with collection.watch(pipeline=change_pipeline) as stream:
 for change in stream:
 print(change)

change_pipeline = { "$match": { "operationType": "update" }},
async with await collection.watch(pipeline=change_pipeline) as stream:
 async for change in stream:
 print(change)

To learn more about modifying your change stream output, see the Modify Change Stream Output section in the MongoDB Server manual.

Modify watch() Behavior

The watch() method accepts optional parameters, which represent options you can use to configure the operation. If you don't specify any options, the driver does not customize the operation.

The following table describes the options you can set to customize the behavior of watch():

database = client["sample_restaurants"]
collection = database["restaurants"]
with collection.watch(full_document='updateLookup') as stream:
 for change in stream:
 print(change)

database = client["sample_restaurants"]
collection = database["restaurants"]
async with await collection.watch(full_document='updateLookup') as stream:
 async for change in stream:
 print(change)

{'_id': {'_data': '...'}, 'operationType': 'update', 'clusterTime': Timestamp(...), 'wallTime': datetime.datetime(...),
 'fullDocument': {'_id': ObjectId('...'), 'address': {...}, 'borough': 'Queens',
 'cuisine': 'Irish', 'grades': [...], 'name': 'Blarney Castle', 'restaurant_id': '40366356'},
 'ns': {'db': 'sample_restaurants', 'coll': 'restaurants'}, 'documentKey': {'_id': ObjectId('...')},
 'updateDescription': {'updatedFields': {'cuisine': 'Irish'}, 'removedFields': [], 'truncatedArrays': []}}

Additional Information

To learn more about change streams, see Change Streams in the MongoDB Server manual.