> For the complete MongoDB documentation index, see www.mongodb.com/docs/llms.txt

# 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](https://www.mongodb.com/docs/manual/core/materialized-views/) 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`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-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`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/addFields.md#mongodb-pipeline-pipe.-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`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/group.md#mongodb-pipeline-pipe.-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`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge.md#mongodb-pipeline-pipe.-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`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge.md#mongodb-pipeline-pipe.-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`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage.

- **Only source changes drive updates.** If the pipeline uses a [`$lookup`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/lookup.md#mongodb-pipeline-pipe.-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](https://kafka.apache.org/) topic and write to [Apache Iceberg](https://iceberg.apache.org/docs/latest/) tables on AWS (Amazon Web Services) S3 (Simple Storage Service). To learn more, see [Apache Kafka Broker](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#std-label-atlas-sp-agg-source-syntax-kafka) and [`$iceberg` Aggregation Stage.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-iceberg.md#std-label-atlas-sp-agg-iceberg)

### Prerequisites

Before you create the stream processor, you must have a [stream processing workspace](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-processing-instance.md#std-label-atlas-sp-manage-spi) with an Atlas connection to the cluster that holds the source data. To add a connection, see [Manage Connections.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-connection-registry.md#std-label-atlas-sp-manage-connections)

Run the commands in this section against your cluster. To connect, see [Connect to a Cluster via mongosh.](https://www.mongodb.com/docs/atlas/mongo-shell-connection.md#std-label-connect-mongo-shell)

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

1. Load the sample dataset.

   This procedure uses the `sales` collection from the [sample\_supplies](https://www.mongodb.com/docs/manual/sample-data/sample-supplies.md#std-label-sample-supplies) dataset. To learn how to load sample data, see [Import Sample Data Into Your Atlas Deployment.](https://www.mongodb.com/docs/atlas/import/load-sample-data.md#std-label-load-sample-data)

2. 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:

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

   **Output:**

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

3. 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:

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

   **Output:**

   ```javascript
   { ok: 1, ... }
   ```

4. Seed the view with the current result.

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

   ```javascript
   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:

   ```javascript
   db.sales_by_channel.find()
   ```

   **Output:**

   ```javascript
   [
     { _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](https://www.mongodb.com/docs/manual/core/materialized-views/). 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.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#std-label-atlas-sp-agg-source-syntax-coll)

### Procedure

Select the Atlas UI or [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh) to create the stream processor.

1. 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.

2. Open the Visual Builder.

   Click Create stream processor.

   Select the Visual Builder.

3. Configure the source.

   In the Source field, select your Atlas connection to the source cluster from the Connection drop-down list.

   In the JSON (Javascript Object Notation) text box, configure the [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage:

   ```json
   {
     "$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`.

4. Add a stage that computes the change delta.

   In the Start building your pipeline pane, click + Custom stage.

   In the JSON (Javascript Object Notation) text box, add an [`$addFields`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/addFields.md#mongodb-pipeline-pipe.-addFields) stage:

   ```json
   {
     "$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.

5. Add a stage that drops events with no effect.

   Click +, then select Custom
   stage.

   In the JSON (Javascript Object Notation) text box, add a [`$match`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/match.md#mongodb-pipeline-pipe.-match) stage:

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

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

6. Add a windowed group stage.

   Click +, then select Custom
   stage.

   In the JSON (Javascript Object Notation) text box, add a [`$tumblingWindow`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-tumbling.md#mongodb-pipeline-pipe.-tumblingWindow) stage:

   ```json
   {
     "$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`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/group.md#mongodb-pipeline-pipe.-group) stage to run inside a window stage.

7. Configure the sink.

   In the Sink field, select your Atlas connection from the Connection drop-down list.

   In the JSON (Javascript Object Notation) text box, configure the [`$merge`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge.md#mongodb-pipeline-pipe.-merge) stage:

   ```json
   {
     "$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

8. 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.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/tier-guide.md#std-label-atlas-sp-tier-guide)

9. Create the stream processor.

   Click Create stream processor.

10. 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.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-stream-processor.md#std-label-atlas-sp-manage-processor)

1) Connect to your stream processing workspace.

   Use the connection string associated with your stream processing workspace to connect using [`mongosh`.](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.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`](https://www.mongodb.com/docs/manual/reference/built-in-roles.md#mongodb-atlasrole-atlasAdmin) role:

   ```sh
   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.

2) Define the source stage.

   Declare a variable and assign it a [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage specification:

   ```javascript
   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`.

3) Define a stage that computes the change delta.

   Declare a variable and assign it an [`$addFields`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/addFields.md#mongodb-pipeline-pipe.-addFields) stage specification:

   ```javascript
   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.

4) Define a stage that drops events with no effect.

   Declare a variable and assign it a [`$match`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/match.md#mongodb-pipeline-pipe.-match) stage specification:

   ```javascript
   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.

5) Define a windowed group stage.

   Declare a variable and assign it a [`$tumblingWindow`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-tumbling.md#mongodb-pipeline-pipe.-tumblingWindow) stage specification:

   ```javascript
   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`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/group.md#mongodb-pipeline-pipe.-group) stage to run inside a window stage.

6) Define the sink stage.

   Declare a variable and assign it a [`$merge`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge.md#mongodb-pipeline-pipe.-merge) stage specification:

   ```javascript
   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

7) Create the stream processor.

   Use the [`sp.createStreamProcessor()`](https://www.mongodb.com/docs/manual/reference/method/sp.createStreamProcessor.md#mongodb-method-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.

   ```javascript
   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.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/tier-guide.md#std-label-atlas-sp-tier-guide)

8) Start the stream processor.

   Use the [`sp.processor.start()`](https://www.mongodb.com/docs/manual/reference/method/sp.processor.start.md#mongodb-method-sp.processor.start) method to start `sales_stats_sp`:

   ```javascript
   sp.sales_stats_sp.start()
   ```

   **Output:**

   ```javascript
   { 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.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-stream-processor.md#std-label-atlas-sp-manage-processor)

### Verify the View Stays Current

Run the commands in this section against your cluster. To connect, see [Connect to a Cluster via mongosh.](https://www.mongodb.com/docs/atlas/mongo-shell-connection.md#std-label-connect-mongo-shell)

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:

```javascript
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`:

```javascript
db.sales_by_channel.find()
```

**Output:**

```javascript
[
  { _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:

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

```javascript
db.sales_by_channel.find()
```

**Output:**

```javascript
[
  { _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](https://github.com/mongodb/ASP_example/tree/main/example_processors/smv) 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:

1. The [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage reads the `support_tickets` change stream with pre- and post-images.

2. The [`$addFields`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/addFields.md#mongodb-pipeline-pipe.-addFields) stage uses a `$switch` on `operationType` to compute `_delta` and extract `_priority` as the group key.

3. The [`$match`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/match.md#mongodb-pipeline-pipe.-match) stage drops events with a delta of zero.

4. The [`$tumblingWindow`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-tumbling.md#mongodb-pipeline-pipe.-tumblingWindow) stage sums the deltas by priority within each one-second window.

5. The [`$merge`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge.md#mongodb-pipeline-pipe.-merge) stage adds each window's delta to the running total, using `lastWindowStart` as a high-water mark to avoid double-counting on replay.

```json
[
  {
    "$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:

```javascript
{ _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:

1. Replace the [`$addFields`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/addFields.md#mongodb-pipeline-pipe.-addFields) stage so that each `$switch` branch returns an `_adjustments` array with one element per affected group key. An escalation returns two elements.

2. Add an [`$unwind`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/unwind.md#mongodb-pipeline-pipe.-unwind) stage and a [`$set`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/set.md#mongodb-pipeline-pipe.-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:

```javascript
// 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.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-stream-processor.md#std-label-atlas-sp-manage-processor)

- To learn about the aggregation stages that Atlas Stream Processing supports, see [Aggregation Pipeline Stages for Stream Processing.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/stream-aggregation-stages.md#std-label-atlas-sp-aggregation)

- To learn about the sources that a stream processor can read, see [`$source` Stage (Stream Processing).](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#std-label-atlas-sp-agg-source)

- To learn about window stages, see [Stream Processor Windows.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/windows.md#std-label-atlas-sp-windows)

- To learn about on-demand materialized views, see [On-Demand Materialized Views.](https://www.mongodb.com/docs/manual/core/materialized-views/)