Build Streaming Materialized Views

A streaming materialized view is a collection that an Atlas Stream Processing stream processor keeps current. To achieve this, a stream processor reads each change to a source collection, computes the effect of that change, and applies the result to the view. The view reflects the current state of the source data without a manual or scheduled refresh.

Streaming materialized views suit workloads that read a computed result more often than the source data changes, such as dashboards, running totals, and open-item counts. The view updates as data arrives, so reads return current results with low latency.

Comparison With On-Demand Materialized Views

MongoDB supports two approaches to materialized views:

An on-demand materialized view stores the result of an aggregation pipeline that you run manually or on a schedule.
A streaming materialized view stores the result of a continuously running Atlas Stream Processing stream processor.

Both approaches store computed results on disk and serve reads directly from the view. They differ in how and when the view updates.

Characteristic	On-Demand Materialized View	Streaming Materialized View
Update trigger	Manual or scheduled	Change-driven, continuous
Latency	Minutes to days	Sub-second to low-second
Data freshness	Point-in-time snapshot	Perpetually synchronized
Compute model	Recomputes full result	Computes incremental effect
Best fit	Batch reporting, periodic aggregation	Live dashboards, operational analytics

Traits of a Streaming Materialized View Processor

Maintaining a streaming materialized view is a pattern that you assemble from standard Atlas Stream Processing aggregation stages. A stream processor that implements this pattern can have the following traits:

Reads a change stream source. The $source stage reads from the source collection with fullDocument and fullDocumentBeforeChange set to required so that the pipeline can compare the state of each document before and after a change.
Computes a signed delta per event. An $addFields stage can use a $switch expression to assign a positive or negative value to each insert, update, or delete based on how the change affects the computed result.
Groups results in a window. Because a stream processor operates on an unbounded stream, every $group stage must run inside a window stage. The windowed $group sums the deltas for each key over the window interval. The window interval also acts as a refresh interval for the view, so it determines how fresh the view is. The smallest interval you can set is one millisecond.
Applies results additively. The $merge stage can use a whenMatched pipeline that adds each window's result to the running total in the view rather than replacing it.
Ends in a sink stage. A stream processor pipeline must end in a sink stage. Use $merge to write to an Atlas collection.

Only incremental aggregations translate to a streaming materialized view. Aggregations that require a full collection scan do not qualify.

Behavior Differences to Consider

A streaming materialized view behaves differently from a batch aggregation in ways that affect downstream consumers:

The view starts from zero. By default, the processor doesn't read pre-existing documents, so seed the view before you start the processor or enable initialSync on the $source stage.
Only source changes drive updates. If the pipeline uses a $lookup stage, later changes to the reference collection do not update documents that the view already wrote.

Create a Streaming Materialized View

The following tutorial creates a streaming materialized view that maintains a count of completed sales by purchase method. The stream processor reads the change stream of the sample_supplies.sales collection and writes to the sample_supplies.sales_by_channel collection.

Note

A stream processor can also read from a Apache Kafka topic and write to Apache Iceberg tables on AWS S3. To learn more, see Apache Kafka Broker and $iceberg Aggregation Stage.

Prerequisites

Before you create the stream processor, you must have a stream processing workspace with an Atlas connection to the cluster that holds the source data. To add a connection, see Manage Connections.

Run the commands in this section against your cluster. To connect, see Connect to a Cluster via mongosh.

Complete the following steps to prepare the source collection and seed the view:

Load the sample dataset.

This procedure uses the sales collection from the sample_supplies dataset. To learn how to load sample data, see Import Sample Data Into Your Atlas Deployment.

Add a status field to the sales collection.

Add a status field so the stream processor can detect completed and returned sales. Run the following command against the cluster:

db.sales.updateMany(
   { status: { $exists: false } },
   { $set: { status: "completed" } }
)

{
  acknowledged: true,
  insertedId: null,
  matchedCount: 5000,
  modifiedCount: 5000,
  upsertedCount: 0
}

Enable change stream pre- and post-images.

Enable pre- and post-images on the source collection so that the stream processor can compute the effect of each change. Run the following command against the cluster:

db.getSiblingDB("sample_supplies").runCommand({
   collMod: "sales",
   changeStreamPreAndPostImages: { enabled: true }
})

{ ok: 1, ... }

Seed the view with the current result.

Run the following batch aggregation against the cluster to populate the view with the current counts:

db.sales.aggregate([
   { $match: { status: "completed" } },
   { $group: { _id: "$purchaseMethod", active_count: { $sum: 1 } } },
   { $merge: {
      into: "sales_by_channel",
      whenMatched: "replace",
      whenNotMatched: "insert"
   } }
])

To confirm the seeded counts, query the view. The sales_by_channel collection contains one document per purchase method:

db.sales_by_channel.find()

[
  { _id: 'Phone', active_count: 596 },
  { _id: 'Online', active_count: 1585 },
  { _id: 'In store', active_count: 2819 }
]

This seed aggregation is itself an on-demand materialized view. The two view types are complementary: you can initialize an on-demand materialized view with a batch aggregation, then start a stream processor that keeps the same collection current. The on-demand view becomes a streaming materialized view.

Note

If your pipeline doesn't use a window stage, you can seed the view with initialSync instead. In this case, the stream processor first ingests every existing document in the source collection as an insert event, then processes new change events. To learn about the $source option initialSync, see MongoDB Collection Change Stream.

Procedure

Select the Atlas UI or mongosh to create the stream processor.

Go to the Stream Processing page for your project.

In the Atlas UI, go to the Stream Processing page for your Atlas project.
Click Manage in the pane of your stream processing workspace that holds your Atlas connection to the source cluster.

Open the Visual Builder.

Click Create stream processor.
Select the Visual Builder.

Configure the source.

In the Source field, select your Atlas connection to the source cluster from the Connection drop-down list.
In the JSON text box, configure the $source stage:

{
  "$source": {
    "connectionName": "<connection-name>",
    "db": "sample_supplies",
    "coll": "sales",
    "config": {
      "fullDocument": "required",
      "fullDocumentBeforeChange": "required"
    }
  }
}

The $source stage opens the change stream on the sample_supplies.sales collection, which is the source of every event the pipeline processes:

fullDocument: "required" includes the current version of the document in every change event.
fullDocumentBeforeChange: "required" includes the document as it existed before the change, which the pipeline needs to detect a sale that changes from completed to returned.

Add a stage that computes the change delta.

In the Start building your pipeline pane, click + Custom stage.
In the JSON text box, add an $addFields stage:

{
  "$addFields": {
    "_delta": {
      "$switch": {
        "branches": [
          {
            "case": {
              "$and": [
                { "$eq": ["$operationType", "insert"] },
                { "$eq": ["$fullDocument.status", "completed"] }
              ]
            },
            "then": 1
          },
          {
            "case": {
              "$and": [
                { "$eq": ["$operationType", "update"] },
                { "$eq": ["$fullDocumentBeforeChange.status", "completed"] },
                { "$eq": ["$fullDocument.status", "returned"] }
              ]
            },
            "then": -1
          },
          {
            "case": {
              "$and": [
                { "$eq": ["$operationType", "delete"] },
                { "$eq": ["$fullDocumentBeforeChange.status", "completed"] }
              ]
            },
            "then": -1
          }
        ],
        "default": 0
      }
    },
    "_channel": {
      "$ifNull": [
        "$fullDocument.purchaseMethod",
        "$fullDocumentBeforeChange.purchaseMethod"
      ]
    }
  }
}

The $addFields stage computes how each event changes the count of completed sales, and identifies the purchase method that it applies to:

The $switch expression assigns a delta of 1 when a sale is inserted as completed, -1 when a sale changes from completed to returned, and -1 when a completed sale is deleted. All other events get a delta of 0.
_channel captures the purchase method from the current document, or from the prior document if the event is a delete.

Add a stage that drops events with no effect.

Click +, then select Custom stage.
In the JSON text box, add a $match stage:

{
  "$match": { "_delta": { "$ne": 0 } }
}

The $match stage drops events with a delta of 0, since these events don't change the running count.

Add a windowed group stage.

Click +, then select Custom stage.
In the JSON text box, add a $tumblingWindow stage:

{
  "$tumblingWindow": {
    "boundary": "processingTime",
    "interval": { "size": 1, "unit": "second" },
    "pipeline": [
      {
        "$group": {
          "_id": "$_channel",
          "active_count": { "$sum": "$_delta" },
          "windowStart": {
            "$first": { "$meta": "stream.window.start" }
          }
        }
      }
    ]
  }
}

The $tumblingWindow stage batches the remaining events so the pipeline writes one summarized count per interval instead of writing on every event:

The window groups events into one-second, non-overlapping intervals based on processing time.
Within each window, the nested $group stage sums the deltas for each purchase method and records when the window started.

Note

A stream processor requires every $group stage to run inside a window stage.

Configure the sink.

In the Sink field, select your Atlas connection from the Connection drop-down list.
In the JSON text box, configure the $merge stage:

{
  "$merge": {
    "into": {
      "connectionName": "<connection-name>",
      "db": "sample_supplies",
      "coll": "sales_by_channel"
    },
    "whenMatched": [
      {
        "$set": {
          "active_count": {
            "$cond": [
              { "$gt": ["$$new.windowStart", { "$ifNull": ["$lastWindowStart", { "$toDate": 0 }] }] },
              { "$add": ["$active_count", "$$new.active_count"] },
              "$active_count"
            ]
          },
          "lastWindowStart": {
            "$max": [
              { "$ifNull": ["$lastWindowStart", { "$toDate": 0 }] },
              "$$new.windowStart"
            ]
          }
        }
      }
    ],
    "whenNotMatched": "insert"
  }
}

The $merge stage writes each window's total to the sample_supplies.sales_by_channel collection, maintaining a running count of completed sales by purchase method:

whenMatched updates active_count only if the incoming window is newer than the last one merged.
whenNotMatched inserts a new document the first time the pipeline sees a purchase method.

Note

The lastWindowStart high-water mark prevents a replayed window from double-counting

Enter processor details.

In the Stream processor name field, enter sales_stats_sp.
Select the tier for the stream processor. To choose a tier for your workload, see Atlas Stream Processing Tier Selection Guide.

Create the stream processor.

Click Create stream processor.

Start the stream processor.

On the Stream Processors tab, select sales_stats_sp and click Start.

The processor now maintains sales_by_channel continuously. To learn more about starting, stopping, and monitoring stream processors, see Develop and Manage Stream Processors.

Connect to your stream processing workspace.

Use the connection string associated with your stream processing workspace to connect using mongosh.

In the pane for your stream processing workspace, click Connect.
In the Connect to your workspace dialog, select the Shell tab.
Copy the connection string displayed in the dialog. It has the following format, where <atlas-stream-processing-url> is the URL of your stream processing workspace and <username> is the username of a database user with the atlasAdmin role:
mongosh "mongodb://<atlas-stream-processing-url>/" --tls --authenticationDatabase admin --username <username> --password <password>
Paste the connection string into your terminal and replace the <password> placeholder with the credentials for the user. Press Enter to connect to your stream processing workspace.

Define the source stage.

Declare a variable and assign it a $source stage specification:

let s = {
  $source: {
    connectionName: "<connection-name>",
    db: "sample_supplies",
    coll: "sales",
    config: {
      fullDocument: "required",
      fullDocumentBeforeChange: "required"
    }
  }
}

The $source stage opens the change stream on the sample_supplies.sales collection, which is the source of every event the pipeline processes:

fullDocument: "required" includes the current version of the document in every change event.
fullDocumentBeforeChange: "required" includes the document as it existed before the change, which the pipeline needs to detect a sale that changes from completed to returned.

Define a stage that computes the change delta.

Declare a variable and assign it an $addFields stage specification:

let d = {
  $addFields: {
    _delta: {
      $switch: {
        branches: [
          {
            case: {
              $and: [
                { $eq: ["$operationType", "insert"] },
                { $eq: ["$fullDocument.status", "completed"] }
              ]
            },
            then: 1
          },
          {
            case: {
              $and: [
                { $eq: ["$operationType", "update"] },
                { $eq: ["$fullDocumentBeforeChange.status", "completed"] },
                { $eq: ["$fullDocument.status", "returned"] }
              ]
            },
            then: -1
          },
          {
            case: {
              $and: [
                { $eq: ["$operationType", "delete"] },
                { $eq: ["$fullDocumentBeforeChange.status", "completed"] }
              ]
            },
            then: -1
          }
        ],
        default: 0
      }
    },
    _channel: {
      $ifNull: [
        "$fullDocument.purchaseMethod",
        "$fullDocumentBeforeChange.purchaseMethod"
      ]
    }
  }
}

The $addFields stage computes how each event changes the count of completed sales, and identifies the purchase method that it applies to:

The $switch expression assigns a delta of 1 when a sale is inserted as completed, -1 when a sale changes from completed to returned, and -1 when a completed sale is deleted. All other events get a delta of 0.
_channel captures the purchase method from the current document, or from the prior document if the event is a delete.

Define a stage that drops events with no effect.

Declare a variable and assign it a $match stage specification:

let f = {
  $match: { _delta: { $ne: 0 } }
}

The $match stage drops events with a delta of 0, since these events don't change the running count.

Define a windowed group stage.

Declare a variable and assign it a $tumblingWindow stage specification:

let w = {
  $tumblingWindow: {
    boundary: "processingTime",
    interval: { size: NumberInt(1), unit: "second" },
    pipeline: [
      {
        $group: {
          _id: "$_channel",
          active_count: { $sum: "$_delta" },
          windowStart: {
            $first: { $meta: "stream.window.start" }
          }
        }
      }
    ]
  }
}

The $tumblingWindow stage batches the remaining events so the pipeline writes one summarized count per interval instead of writing on every event:

The window groups events into one-second, non-overlapping intervals based on processing time.
Within each window, the nested $group stage sums the deltas for each purchase method and records when the window started.

Note

A stream processor requires every $group stage to run inside a window stage.

Define the sink stage.

Declare a variable and assign it a $merge stage specification:

let m = {
  $merge: {
    into: {
      connectionName: "<connection-name>",
      db: "sample_supplies",
      coll: "sales_by_channel"
    },
    whenMatched: [
      {
        $set: {
          active_count: {
            $cond: [
              { $gt: ["$$new.windowStart", { $ifNull: ["$lastWindowStart", { $toDate: 0 }] }] },
              { $add: ["$active_count", "$$new.active_count"] },
              "$active_count"
            ]
          },
          lastWindowStart: {
            $max: [
              { $ifNull: ["$lastWindowStart", { $toDate: 0 }] },
              "$$new.windowStart"
            ]
          }
        }
      }
    ],
    whenNotMatched: "insert"
  }
}

The $merge stage writes each window's total to the sample_supplies.sales_by_channel collection, maintaining a running count of completed sales by purchase method:

whenMatched updates active_count only if the incoming window is newer than the last one merged.
whenNotMatched inserts a new document the first time the pipeline sees a purchase method.

Note

The lastWindowStart high-water mark prevents a replayed window from double-counting

Create the stream processor.

Use the sp.createStreamProcessor() method to create a stream processor named sales_stats_sp from the stages you defined. The $group stage belongs to the nested pipeline of the $tumblingWindow stage, so don't include it in the processor pipeline definition.

sp.createStreamProcessor("sales_stats_sp", [s, d, f, w, m])

To assign the processor to a tier other than the tier of your stream processing workspace, pass a tier option in the options object. To choose a tier for your workload, see Atlas Stream Processing Tier Selection Guide.

Start the stream processor.

Use the sp.processor.start() method to start sales_stats_sp:

sp.sales_stats_sp.start()

{ ok: 1 }

The processor now maintains sales_by_channel continuously. To learn more about starting, stopping, and monitoring stream processors, see Develop and Manage Stream Processors.

Verify the View Stays Current

Run the commands in this section against your cluster. To connect, see Connect to a Cluster via mongosh.

After you start the processor, changes to the sales collection update sales_by_channel within seconds. To confirm this, insert a new completed online sale:

db.sales.insertOne({
   saleDate: new Date(),
   purchaseMethod: "Online",
   status: "completed",
   items: [],
   customer: {},
   couponUsed: false
})

The processor increments the Online count and records the window boundary in lastWindowStart:

db.sales_by_channel.find()

[
  { _id: 'Phone', active_count: 596 },
  {
    _id: 'Online',
    active_count: 1586,
    lastWindowStart: ISODate('2026-07-23T15:18:01.000Z')
  },
  { _id: 'In store', active_count: 2819 }
]

When a customer returns that sale, the Online count returns to its seeded value:

db.sales.updateOne(
   { purchaseMethod: "Online", status: "completed" },
   { $set: { status: "returned" } }
)

db.sales_by_channel.find()

[
  { _id: 'Phone', active_count: 596 },
  {
    _id: 'Online',
    active_count: 1585,
    lastWindowStart: ISODate('2026-07-23T15:18:13.000Z')
  },
  { _id: 'In store', active_count: 2819 }
]

Examples

These examples walk through the smv example in the Atlas Stream Processing example repository. It maintains a queue_stats collection that holds a running count of open support tickets, with one document per priority level. The stream processor reads the support_tickets change stream and updates each count as tickets are opened, resolved, and deleted.

The pipeline adjusts the count for a ticket's priority by +1 when a ticket opens and -1 when a ticket is resolved or deleted. The aggregation has five stages:

The $source stage reads the support_tickets change stream with pre- and post-images.
The $addFields stage uses a $switch on operationType to compute _delta and extract _priority as the group key.
The $match stage drops events with a delta of zero.
The $tumblingWindow stage sums the deltas by priority within each one-second window.
The $merge stage adds each window's delta to the running total, using lastWindowStart as a high-water mark to avoid double-counting on replay.

[
  {
    "$source": {
      "connectionName": "<connection-name>",
      "db": "support",
      "coll": "support_tickets",
      "config": {
        "fullDocument": "required",
        "fullDocumentBeforeChange": "required"
      }
    }
  },
  {
    "$addFields": {
      "_delta": {
        "$switch": {
          "branches": [
            {
              "case": {
                "$and": [
                  { "$eq": ["$operationType", "insert"] },
                  { "$eq": ["$fullDocument.status", "open"] }
                ]
              },
              "then": 1
            },
            {
              "case": {
                "$and": [
                  { "$eq": ["$operationType", "update"] },
                  { "$eq": ["$fullDocumentBeforeChange.status", "open"] },
                  { "$eq": ["$fullDocument.status", "resolved"] }
                ]
              },
              "then": -1
            },
            {
              "case": {
                "$and": [
                  { "$eq": ["$operationType", "delete"] },
                  { "$eq": ["$fullDocumentBeforeChange.status", "open"] }
                ]
              },
              "then": -1
            }
          ],
          "default": 0
        }
      },
      "_priority": {
        "$ifNull": [
          "$fullDocument.priority",
          "$fullDocumentBeforeChange.priority"
        ]
      }
    }
  },
  { "$match": { "_delta": { "$ne": 0 } } },
  {
    "$tumblingWindow": {
      "boundary": "processingTime",
      "interval": { "size": 1, "unit": "second" },
      "pipeline": [
        {
          "$group": {
            "_id": "$_priority",
            "open_count": { "$sum": "$_delta" },
            "windowStart": {
              "$first": { "$meta": "stream.window.start" }
            }
          }
        }
      ]
    }
  },
  {
    "$merge": {
      "into": {
        "connectionName": "<connection-name>",
        "db": "support",
        "coll": "queue_stats"
      },
      "whenMatched": [
        {
          "$set": {
            "open_count": {
              "$cond": [
                { "$gt": ["$$new.windowStart", { "$ifNull": ["$lastWindowStart", { "$toDate": 0 }] }] },
                { "$add": ["$open_count", "$$new.open_count"] },
                "$open_count"
              ]
            },
            "lastWindowStart": {
              "$max": [
                { "$ifNull": ["$lastWindowStart", { "$toDate": 0 }] },
                "$$new.windowStart"
              ]
            }
          }
        }
      ],
      "whenNotMatched": "insert"
    }
  }
]

Each document in the queue_stats collection resembles the following:

{ _id: "P1", open_count: <num>, lastWindowStart: <timestamp> }

When one event must adjust two group keys, replace the scalar _delta with an _adjustments array and fan the array out into one document per adjustment. Modify the single-key pipeline as follows:

Replace the $addFields stage so that each $switch branch returns an _adjustments array with one element per affected group key. An escalation returns two elements.
Add an $unwind stage and a $set stage after $addFields. The $unwind stage splits each event into one document per adjustment and drops empty arrays, which replaces the $match stage. The $set stage promotes the _adjustments fields to top-level _priority and _delta so that the remaining stages work unchanged.

Replace stages 2 and 3 of the single-key pipeline with the following:

// Stage 2 (replacement): Compute an _adjustments array.
{
  $addFields: {
    _adjustments: {
      $switch: {
        branches: [
          {
            case: { $and: [
              { $eq: ["$operationType", "insert"] },
              { $eq: ["$fullDocument.status", "open"] }
            ]},
            then: [{ _priority: "$fullDocument.priority", _delta: 1 }]
          },
          {
            case: { $and: [
              { $eq: ["$operationType", "update"] },
              { $eq: ["$fullDocumentBeforeChange.status", "open"] },
              { $eq: ["$fullDocument.status", "resolved"] }
            ]},
            then: [{ _priority: "$fullDocumentBeforeChange.priority",
              _delta: -1 }]
          },
          {
            case: { $and: [
              { $eq: ["$operationType", "update"] },
              { $eq: ["$fullDocument.status", "open"] },
              { $ne: ["$fullDocument.priority",
                "$fullDocumentBeforeChange.priority"] }
            ]},
            then: [
              { _priority: "$fullDocumentBeforeChange.priority",
                _delta: -1 },
              { _priority: "$fullDocument.priority", _delta:  1 }
            ]
          },
          {
            case: { $and: [
              { $eq: ["$operationType", "delete"] },
              { $eq: ["$fullDocumentBeforeChange.status", "open"] }
            ]},
            then: [{ _priority: "$fullDocumentBeforeChange.priority",
              _delta: -1 }]
          }
        ],
        default: []
      }
    }
  }
},
// Stage 3 (replacement): Fan out into one document per adjustment,
// then lift the adjustment fields back to the top level.
{ $unwind: "$_adjustments" },
{
  $set: {
    _priority: "$_adjustments._priority",
    _delta: "$_adjustments._delta"
  }
}

Additional Information

To learn more about the stages and concepts in this guide, see the following resources:

To learn how to create, start, stop, and monitor stream processors, see Develop and Manage Stream Processors.
To learn about the aggregation stages that Atlas Stream Processing supports, see Aggregation Pipeline Stages for Stream Processing.
To learn about the sources that a stream processor can read, see $source Stage (Stream Processing).
To learn about window stages, see Stream Processor Windows.
To learn about on-demand materialized views, see On-Demand Materialized Views.

Back

Tier Sizing Guide

Configure Multi-Collection Initial Sync

Build Streaming Materialized Views

Comparison With On-Demand Materialized Views

Traits of a Streaming Materialized View Processor

Behavior Differences to Consider

Create a Streaming Materialized View

Prerequisites

Load the sample dataset.

Add a status field to the sales collection.

Enable change stream pre- and post-images.

Seed the view with the current result.

Procedure

Go to the Stream Processing page for your project.

Open the Visual Builder.

Configure the source.

Add a stage that computes the change delta.

Add a stage that drops events with no effect.

Add a windowed group stage.

Configure the sink.

Enter processor details.

Create the stream processor.

Start the stream processor.

Connect to your stream processing workspace.

Define the source stage.

Define a stage that computes the change delta.

Define a stage that drops events with no effect.

Define a windowed group stage.

Define the sink stage.

Create the stream processor.

Start the stream processor.

Verify the View Stays Current

Examples

Count open tickets by priorityCompute a signed delta per event and add it to the count for each priority.

Adjust multiple group keys from a single eventFan a single event out into one adjustment per affected group key.

Additional Information