The $changeStream stage turns an aggregation pipeline into a live feed of database changes—inserts, updates, deletes, and more—so apps can react instantly instead of polling.
Watch collection changes.
Pipeline must start here.
insert, update, delete…
Whole doc on updates.
Survive reconnects.
Preferred in mongosh.
$changeStream is an aggregation stage that returns a tailable change stream cursor instead of a one-shot result set. Each event describes what changed, when, and on which namespace (ns.db, ns.coll).
Change streams need a replica set or sharded cluster—they do not work on a default standalone mongod. For local dev, convert to a single-member replica set. The cursor blocks until a change occurs (or you close it).
Use $changeStream for live notifications, cache invalidation, ETL triggers, and sync to search indexes. In mongosh, db.collection.watch() is usually easier—it builds a pipeline starting with $changeStream automatically.
$changeStream accepts an options object (all fields optional):
{
$changeStream: {
fullDocument: "default" | "updateLookup" | "whenAvailable" | "required",
fullDocumentBeforeChange: "off" | "whenAvailable" | "required",
resumeAfter: <resume token document>,
startAfter: <resume token document>,
startAtOperationTime: <timestamp>,
showExpandedEvents: <boolean>,
allChangesForCluster: <boolean>
}
}$changeStream must be stage 1 in the pipeline (or use watch())..hasNext() / .next() in a loop; the stream stays open until closed._id — resume token; save it to continue after disconnect with resumeAfter.fullDocument — updateLookup fetches the post-update document for update events.resumeAfter, startAfter, and startAtOperationTime are mutually exclusive.$match can filter events; avoid $project that drops _id if you need resume.const cursor = db.orders.watch([
{ $changeStream: {} }
]);
// Block until next change, then print event
printjson(cursor.next());| Question | Answer |
|---|---|
| Stage position | Must be first in the pipeline |
| Deployment | Replica set or sharded cluster required |
| Common helper | db.coll.watch([ … ]) |
| Key event fields | operationType, fullDocument, documentKey, ns |
| Update full doc | fullDocument: "updateLookup" |
| Resume | Pass last event _id as resumeAfter |
db.orders.watch([
{ $changeStream: {} }
])All changes
db.orders.watch([
{ $changeStream: {} },
{ $match: {
operationType: "insert"
}}
])Filter events
{ $changeStream: {
fullDocument: "updateLookup"
}}Post-update snapshot
{ $changeStream: {
resumeAfter: token
}}After disconnect
Options inside the $changeStream object customize the stream:
fullDocument OptionalControls whether update events include the full document. default omits it; updateLookup loads the current document.
fullDocument: "updateLookup"resumeAfter OptionalResume token (_id from a prior event) to continue where you left off. Cannot resume after an invalidate event.
resumeAfter: lastEvent._idstartAfter OptionalLike resumeAfter but can open a new stream after invalidation (e.g. collection recreated).
startAfter: savedTokenstartAtOperationTime OptionalStart at a cluster time (BSON timestamp). Useful when you know when to begin listening.
startAtOperationTime: Timestamp(1, 0)Sample orders collection—open a stream, read change events, filter inserts, fetch full documents on updates, and resume after a token.
Replica set requirement and sample collection.
// Change streams require replica set or sharded cluster.
// Verify:
rs.status()
db.orders.drop()
db.orders.insertOne({
product: "Keyboard",
status: "pending",
amount: 45
})Run rs.status() to confirm replica set mode. Insert seed data, then open a change stream in another shell session to observe new writes.
// Session A — open the stream
const stream = db.orders.watch([
{ $changeStream: {} }
])
// Session B — make a change
db.orders.insertOne({
product: "Mouse",
status: "pending",
amount: 25
})
// Session A — read one event
const event = stream.next()
printjson(event)
/* Sample insert event:
{
_id: { _data: "8262…" }, // resume token
operationType: "insert",
fullDocument: { product: "Mouse", status: "pending", amount: 25, _id: … },
ns: { db: "test", coll: "orders" },
documentKey: { _id: … },
clusterTime: Timestamp(…),
wallTime: ISODate("…")
} */
watch() wraps an aggregation whose first stage is $changeStream. The cursor waits until MongoDB commits a change, then returns one event document.
Filter events, load full documents, and resume streams.
const stream = db.orders.watch([
{ $changeStream: {} },
{
$match: {
operationType: { $in: ["insert"] }
}
}
])
// Updates and deletes are filtered out server-side
db.orders.updateOne(
{ product: "Mouse" },
{ $set: { status: "shipped" } }
)
// → no event on this stream
db.orders.insertOne({ product: "Monitor", amount: 180 })
// → stream.next() returns the insert event
Stages after $changeStream process each event like a normal pipeline. $match on operationType is the most common filter for ETL jobs that only care about new rows.
const stream = db.orders.watch([
{
$changeStream: {
fullDocument: "updateLookup"
}
}
])
db.orders.updateOne(
{ product: "Keyboard" },
{ $set: { status: "shipped", amount: 42 } }
)
const event = stream.next()
// event.operationType === "update"
// event.fullDocument → entire document AFTER update
// event.updateDescription → { updatedFields: { status, amount }, … }
printjson({
op: event.operationType,
status: event.fullDocument.status,
amount: event.fullDocument.amount
})
By default, updates may only include updateDescription (changed fields). updateLookup performs a lookup so consumers get the full current document—handy for search sync and cache refresh.
// First connection
let stream = db.orders.watch([{ $changeStream: {} }])
let lastToken = null
while (stream.hasNext()) {
const event = stream.next()
lastToken = event._id // save resume token
processEvent(event) // your app logic
if (shouldStop) break
}
// Later — reconnect without missing events
stream = db.orders.watch([
{
$changeStream: {
resumeAfter: lastToken
}
}
])
// Alternative: aggregate() with $changeStream (same idea)
// db.orders.aggregate([
// { $changeStream: { resumeAfter: lastToken } }
// ])
Every change event includes a resume token in _id. Persist it (database, Redis, file) so workers survive restarts. Use startAfter when the old stream was invalidated.
$changeStream WorksClient runs watch() or aggregate([{ $changeStream }]). MongoDB registers interest in the namespace.
A write hits the collection. The oplog (replica set) records the operation.
MongoDB builds a change event with operationType, optional fullDocument, and resume token _id.
Your loop processes the event—update cache, push to WebSocket, enqueue a job.
mongod returns an error when opening a change stream.operationType values include insert, update, replace, delete, invalidate, and (with showExpandedEvents) DDL events.documentKey but usually no fullDocument.invalidate event closes the stream (e.g. collection dropped)—use startAfter to reopen.$changeStream is MongoDB’s built-in way to subscribe to data changes. Open with watch(), filter with $match, enable updateLookup when you need full documents, and persist resume tokens for reliability.
Remember the replica set requirement before debugging “change stream not supported” errors. Next in the series: $collStats for collection statistics.
db.collection.watch() over raw aggregate() for resume support$match early to reduce network traffic to your appfullDocument: "updateLookup" when downstream needs the whole docinvalidate and reconnect with startAfter_id in $project if you rely on automatic resumefind() when push-based $changeStream fits the use case$changeStreamUse these points when building reactive MongoDB apps.
Push, not poll.
BasicsPipeline rule.
SyntaxRequired deploy.
SetupFull doc on update.
Optionevent._id saved.
ReliabilityChange streams were introduced in MongoDB 3.6 and largely replaced manual oplog tailing. You can also open a database-level stream with db.watch() or cluster-wide on admin with allChangesForCluster. See the official $changeStream docs.
Subscribe to live updates with $changeStream, then inspect collections with $collStats.
5 people found this page helpful