Change stream pipelines for filtering in MongoDB - Mini Project: Build & Apply

Change Stream Pipelines for Filtering in MongoDB

📖 Scenario: You are managing a MongoDB database for a small online bookstore. You want to monitor changes to the orders collection, but only for orders where the status is "shipped". This helps you react quickly to shipped orders without processing all changes.

🎯 Goal: Build a MongoDB change stream pipeline that filters change events to only include those where the status field in the fullDocument is "shipped".

📋 What You'll Learn

Create a change stream pipeline array called pipeline with a match stage filtering fullDocument.status equal to "shipped".

Create a variable called changeStream that watches the orders collection using the pipeline.

Use the watch method with the pipeline to filter change events.

Add a final option to the watch method to include the full document in the change event.

💡 Why This Matters

🌍 Real World

Filtering change streams is useful in real-time applications like order tracking, notifications, and syncing data where only specific changes matter.

💼 Career

Understanding change stream pipelines is important for backend developers and database administrators working with MongoDB to build efficient, reactive systems.

1

Create the change stream pipeline array

Create a variable called pipeline and set it to an array containing one object. This object should be a $match stage that filters documents where fullDocument.status is exactly "shipped".

MongoDB

# Your code here

Need a hint?

Use $match inside the array to filter change events by the status field inside fullDocument.

2

Create the change stream watcher

Create a variable called changeStream and assign it the result of calling db.orders.watch(pipeline).

MongoDB

pipeline = [
    { "$match": { "fullDocument.status": "shipped" } }
]
# Your code here

Need a hint?

Use the watch method on db.orders with the pipeline variable.

3

Add option to include full document in change events

Modify the changeStream assignment to call db.orders.watch(pipeline, { fullDocument: "updateLookup" }) so that the full updated document is included in change events.

MongoDB

pipeline = [
    { "$match": { "fullDocument.status": "shipped" } }
]
# Your code here

Need a hint?

Pass the option { fullDocument: "updateLookup" } as the second argument to watch.

4

Complete the change stream setup

Ensure the pipeline and changeStream variables are defined exactly as before, with the watch method using the pipeline and the option { fullDocument: "updateLookup" }.

MongoDB

pipeline = [
    { "$match": { "fullDocument.status": "shipped" } }
]
changeStream = db.orders.watch(pipeline, { fullDocument: "updateLookup" })
# Your code here

Need a hint?

Double-check that the pipeline and changeStream variables are exactly as required.