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 in the Atlas UI

The following procedure 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.

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

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 to read the sales collection with pre- and post-images:

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

Add a stage that computes the change delta for each event.

In the Start building your pipeline pane, click + Custom stage.
In the JSON text box, add an $addFields stage that assigns a signed delta to each insert, update, or delete and captures the purchase method:

{
  "$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"
      ]
    }
  }
}

Add a stage that drops events with no effect.

Click +, then select Custom stage.
In the JSON text box, add a $match stage that removes events with a delta of zero:

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

Add a windowed group stage.

Click +, then select Custom stage.
In the JSON text box, add a $tumblingWindow stage that sums the deltas by purchase method over one-second intervals:

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

Important

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 to add each window's result to the running total in sales_by_channel:

{
  "$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"
  }
}

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 Stream Processors.

Verify the View Stays Current

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.

Count open tickets by priority

Compute a signed delta per event and add it to the count for each priority.

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> }

Adjust multiple group keys from a single event

Fan a single event out into one adjustment per affected group key.

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 Stream Processors.
To learn about the aggregation stages that Atlas Stream Processing supports, see Aggregation Pipeline Stages.
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

Architecture