Change stream pipelines for filtering in MongoDB

Introduction

Change stream pipelines let you watch only the database changes you care about. This saves time and resources by filtering out unwanted updates.

You want to track only new orders in an online store.

You need to monitor changes to user profiles but ignore other data.

You want to react only when a document is deleted.

You want to log changes where a specific field value changes.

You want to update a cache only when certain data changes.

Syntax

MongoDB

db.collection.watch([pipeline], { options })

The pipeline is an array of aggregation stages that filter or transform change events.

You can use stages like $match to filter events by operation type or document fields.

Examples

Watch only new documents inserted into the orders collection.

MongoDB

db.orders.watch([
  { $match: { operationType: 'insert' } }
])

Watch changes where the user's status is 'active'.

MongoDB

db.users.watch([
  { $match: { 'fullDocument.status': 'active' } }
])

Watch only updates or replacements in the products collection.

MongoDB

db.products.watch([
  { $match: { operationType: { $in: ['update', 'replace'] } } }
])

Sample Program

This watches the orders collection and prints a message only when a new order is inserted.

MongoDB

const changeStream = db.orders.watch([
  { $match: { operationType: 'insert' } }
]);

changeStream.on('change', (change) => {
  print('New order inserted with id:', change.fullDocument._id);
});

Important Notes

Change streams require a replica set or a sharded cluster in MongoDB.

Filtering early in the pipeline reduces the amount of data sent to your application.

You can combine multiple stages to filter by operation type and document fields.

Summary

Change stream pipelines filter database change events.

Use $match to watch only specific operations or documents.

Filtering saves resources and focuses on relevant changes.