Use Change Streams

Your application is built, deployed, and users are interacting with it. Then it happens again — a record changes, a new order lands, a user updates their profile, and you have no idea. Polling feels clunky, webhooks require external infrastructure, and your database remains a silent black box. MongoDB's change streams solve this by giving you a real-time, ordered, and resumable feed of data changes directly from the database. In this lesson, you'll learn how to use change streams to react to data changes, turning your MongoDB deployment from a passive store into an active event source.

1. The problem this lesson solves

Traditional ways to detect data changes are reactive by nature and full of trade-offs. Polling, for example, queries the database every few seconds looking for new or updated documents. It's simple but wasteful — you're constantly hammering the database for data that may not exist. It also introduces latency: a change that happens one millisecond after a poll isn't seen until the next poll, and the database load scales with every client that polls.

Webhooks and message queues (like RabbitMQ or Apache Kafka) solve the latency problem, but they add external infrastructure. You now have to operate a separate system, handle network failures, and write code to bridge your database and that system. Worse, they depend on your application to propagate changes — if your app crashes mid-write, the notification is lost.

Change streams flip the model. Instead of your application asking "what changed?", the database tells you what changed. MongoDB watches collections, databases, or entire deployments and emits a change event — a document that describes the operation (insert, update, delete, replace) and the affected data — in near real time. Your code stays simple, your infrastructure stays lean, and your reaction time drops to milliseconds.

This lesson is part of the MongoDB track, and it builds directly on your understanding of collections, queries, and the aggregation pipeline. By the end, you'll be able to wire your MongoDB cluster into your application's event-driven architecture.

2. Core concept / mental model

Think of a change stream as a live subscription to your data. It's like a river: documents flow downstream, and you stand on the bank with a net that catches every change that floats by. You don't need to pull the river toward you (polling); you just watch, and the river brings you what you need.

Behind the scenes, change streams use the MongoDB oplog (operations log), a capped collection that records every write operation. When you open a change stream, MongoDB reads the oplog and streams changes to your client in the order they occurred. Because the oplog is ordered and durable, you get a guarantee that change events are delivered in the same order as the writes — critical for features like inventory management or financial transactions.

Key terms you'll encounter:

• Change event: A document that describes a single database change. It includes operationType, ns (namespace: database and collection), documentKey, and fullDocument (for inserts and replaces).
• Resume token: A unique identifier for a point in the change stream. You can save it and resume from that point later, even after a client crash.
• Pipeline: Change streams accept an aggregation pipeline, allowing you to filter and transform events before they reach your application.

Change streams are available in replica sets and sharded clusters (including MongoDB Atlas), but not in standalone instances. This is because the oplog only exists on replica sets. If you're on a standalone server, you'll need to convert it to a single-node replica set — we'll cover that in the hands-on section.

3. How it works step by step

Step 1: Ensure your deployment is change-stream-ready

Change streams require a replica set or a sharded cluster. MongoDB Atlas clusters (M0 and above) are replica sets by default, so they work out of the box. For a local MongoDB instance, you'll need to initialize it as a single-node replica set.

Step 2: Open a change stream

From your application, you open a stream on a collection, a database, or the entire cluster. The most common approach is collection-level:

import pymongo

client = pymongo.MongoClient("mongodb://localhost:27017/")
db = client["shop"]
orders = db["orders"]

# Open a change stream on the 'orders' collection
with orders.watch() as stream:
    for change in stream:
        print(change)
        # React to the change
        process_order(change)

This blocks and yields change events as they occur. The watch() method returns a CommandCursor that you can iterate over.

Step 3: Filter and transform with an aggregation pipeline

You can pass a pipeline to watch() to only receive events you care about. For example, you might want to ignore all inserts and only watch updates to the status field:

pipeline = [
    {"$match": {"operationType": "update", "updateDescription.updatedFields.status": {"$exists": True}}}
]
with orders.watch(pipeline=pipeline) as stream:
    for change in stream:
        print(change)

Step 4: Handle resume tokens

Change streams are resumable. If your application crashes, you can restart the stream from the last processed token, avoiding duplicate processing or missed events. The resume_token field in each change event contains the necessary data.

resume_token = None
with orders.watch() as stream:
    for change in stream:
        process(change)
        resume_token = change["_id"]  # save this

# Later, resume from that token
with orders.watch(resume_after=resume_token) as stream:
    for change in stream:
        process(change)

Step 5: React in your application

The change event is a JSON document. You can trigger any action: update a cache, send a notification, synchronize a search index, or feed an analytics dashboard. The key is that you decide what "react" means for your use case.

4. Hands-on walkthrough

Now let's put this into practice. We'll set up a local environment, enable change streams, and build a small Python script that reacts to new orders.

Create a replica set (if needed)

If you have a standalone MongoDB, convert it to a single-node replica set by starting mongod with the --replSet flag and initializing it:

# Start mongod as a single-node replica set
mongod --dbpath /data/db --replSet rs0

# In another terminal, connect and initialize
mongosh --eval "rs.initiate()"

You should see output like { "ok": 1 }.

Build a change-stream listener

Create a file watch_orders.py:

import pymongo
import json

client = pymongo.MongoClient("mongodb://localhost:27017/")
db = client["shop"]
orders = db["orders"]

# Open a change stream that only listens for inserts
pipeline = [{"$match": {"operationType": "insert"}}]

print("Listening for new orders...")
with orders.watch(pipeline=pipeline) as stream:
    for change in stream:
        order = change["fullDocument"]
        print(f"New order received: {json.dumps(order, indent=2)}")
        # Example reaction: log and send to a shipping service
        print(f"Shipping order {order['_id']} to {order['customer']}")

Now, open a separate terminal and insert a document into the orders collection:

mongosh --eval 'db.orders.insertOne({customer: "Alice", total: 129.99, status: "pending"})' shop

In your listener terminal, you'll see output similar to:

Listening for new orders...
New order received: {
  "_id": "...",
  "customer": "Alice",
  "total": 129.99,
  "status": "pending"
}
Shipping order ObjectId('...') to Alice

The change event arrived nearly instantly, with no polling.

React to updates with a resume token

Let's build a more robust example that resumes after a crash. We'll save the resume token to a file after each event:

import pymongo
import json
import os

client = pymongo.MongoClient("mongodb://localhost:27017/")
orders = client["shop"]["orders"]

TOKEN_FILE = "resume_token.json"

def load_token():
    if os.path.exists(TOKEN_FILE):
        with open(TOKEN_FILE) as f:
            return json.load(f)
    return None

def save_token(token):
    with open(TOKEN_FILE, "w") as f:
        json.dump(token, f)

resume_token = load_token()
opts = {"resume_after": resume_token} if resume_token else {}

with orders.watch(**opts) as stream:
    for change in stream:
        print("Change:", change)
        # Process the change...
        save_token(change["_id"])

Now if your script crashes, it will resume from the last processed event on the next run, avoiding duplicates.

5. Compare options / when to choose what

Change streams aren't the only way to react to data changes. Here's a comparison with other common approaches:

When to choose change streams: - You need sub-second reaction time. - You want to avoid extra infrastructure (no Kafka or RabbitMQ). - You're building a microservice or a real-time feature (e.g., live dashboard, cache invalidation, search index sync).

When to consider alternatives: - If you're already running Kafka for event streaming and need multi-source aggregation, a message queue may be a better fit. - If you need to notify external systems over HTTP, webhooks are more direct (though you can bridge change streams to webhooks yourself).

Variations

• Full document lookup: For update operations, change events don't include the entire document by default. You can set full_document="updateLookup" to fetch the latest version in the event.
• Database/cluster-wide streams: Instead of collection-level, use client.watch() for all collections in the cluster, or db.watch() for all collections in a database.
• Pre/post-images: In MongoDB 6.0+, you can configure streams to include the document before and/or after an update, which is handy for auditing.

6. Troubleshooting & edge cases

"Change streams are not supported on standalone instances"

Error: pymongo.errors.OperationFailure: The $changeStream stage is only supported on replica sets

Fix: Convert your instance to a replica set as shown in the hands-on section. In Atlas, ensure you're not on a free M0 cluster? Actually, M0 supports change streams, but you must wait a few seconds after the cluster is created.

No events after inserting a document

• Cause: Your change stream pipeline filters out the operation type.
• Fix: Check your $match condition. If you're filtering for operationType: "insert", your watch() pipeline must include that condition at the start. Also, verify that the change event's operationType is exactly what you expect (e.g., 'update' for updates).
• Hidden gotcha: In Python, the $match stage must be the first stage in the pipeline. If you put $project first, it will fail.

Duplicate events after a crash

• Cause: Your resume token wasn't saved, or you resumed from a token that's older than the last processed event.
• Fix: Save the _id of each change after processing it, not before. If you process and then crash, you may miss an event; if you save before processing, you may process twice on duplicate events. Use the resume token to restart exactly from the last saved event.

Full document is missing for updates

• Cause: By default, change events for updates include only the fields that changed.
• Fix: Pass full_document="updateLookup" to the watch() method to include the full document. Note that for deletes, no fullDocument is included (you have to store the document key).

Stream stops silently in production

• Change streams time out after 10 seconds of inactivity (the maxAwaitTimeMS default). Your application should handle ServerSelectionTimeoutError and resume. In Python, the with orders.watch(...) block will raise an exception if the connection is lost; you need to catch it and start a new stream with the last resume token.

Pro tip: Always use a resume token for any production change stream. It's the difference between a robust event-driven system and a flaky one.

7. What you learned & what's next

You now understand how to use change streams to react to data changes. Let's recap the key points:

• Core idea: Change streams provide a real-time, ordered, resumable feed of database changes via the oplog.
• Practical application: You can open a stream, filter with an aggregation pipeline, and react to events in your application with milliseconds of latency.
• Resume tokens: They ensure you never miss an event, even after a crash.
• Comparison: Change streams are often simpler and more lightweight than polling or external message queues.
• Troubleshooting: You know common errors like standalone instances, pipeline order, and full document vs. update operations.

You've covered learning objective 1 (explain the core idea) and objective 2 (complete a practical exercise). You're now ready to integrate change streams into your own projects.

Next in the MongoDB track, you'll learn how to build a real-time application with MongoDB — combining change streams with a Node.js or Python web server, and you'll see how to scale your event-driven architecture as your data grows. Keep your resume token handy — you'll reuse that pattern in the next lesson.