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

<!--
Tab options on this page. Append to the .md URL to filter:
  ?tabs=<id,...>   select specific tabs (e.g. ?tabs=nodejs,shell)
  ?allTabs=true    include every tab
  (no param)       default: one tab per tabset

Available tabs:
  other tabs: visual-builder, json-editor
-->

# Develop and Manage Stream Processors

An Atlas Stream Processing stream processor applies the logic of a uniquely named [stream aggregation pipeline](https://www.mongodb.com/docs/atlas/atlas-stream-processing/stream-aggregation-stages.md#std-label-atlas-sp-aggregation) to your streaming data. Atlas Stream Processing saves each stream processor definition to persistent storage so that it can be reused. You can only use a given stream processor in the [stream processing workspace](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-processing-instance.md#std-label-atlas-sp-manage-spi) in which its definition is stored.

To let Atlas Stream Processing adjust a processor's tier automatically in response to resource usage, enable [vertical autoscaling.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/autoscaling.md#std-label-atlas-sp-autoscaling)

## Prerequisites

To create and manage a 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)

- A database user with the [`atlasAdmin`](https://www.mongodb.com/docs/manual/reference/built-in-roles.md#mongodb-atlasrole-atlasAdmin) role to create and run stream processors

- An Atlas cluster

## Considerations

Many stream processor commands require you to specify the name of the relevant stream processor in the method invocation. The syntax described in the following sections assumes strictly alphanumeric names. If your stream processor's name includes non-alphanumeric characters such as hyphens (`-`) or full stops (`.`), you must enclose the name in square brackets (`[]`) and double quotes (`""`) in the method invocation, as in `sp["special-name-stream"].stats()`.

## Create a Stream Processor Interactively

You cannot create a stream processor interactively using the Atlas Administration API. To learn how to create a stream processor interactively using [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh), select MongoDB Shell from the Interface dropdown.

You cannot create a stream processor interactively using the Atlas UI. To learn how to create a stream processor interactively using [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh), select MongoDB Shell from the Interface dropdown.

You cannot create a stream processor interactively using the Atlas CLI. To learn how to create a stream processor interactively using [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh), select MongoDB Shell from the Interface dropdown.

You can create a stream processor interactively with the [`sp.process()`](https://www.mongodb.com/docs/manual/reference/method/sp.process.md#mongodb-method-sp.process) method in [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh). Stream processors that you create interactively exhibit the following behavior:

- Write output and dead letter queue documents to the shell

- Begin running immediately upon creation

- Run until one of the following occurs: They return the number of documents set by the `options.limit` field, 10 minutes elapse, or the user stops them

- Don't persist after stopping

Stream processors that you create interactively are intended for prototyping. To create a persistent stream processor, see [Create a Stream Processor.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-stream-processor.md#std-label-streams-manage-create)

`sp.process()` has the following syntax:

```sh
sp.process(<pipeline>, <options>)
```

| Field | Type | Necessity | Description |
| --- | --- | --- | --- |
| `pipeline` | array | Required | [Stream aggregation pipeline](https://www.mongodb.com/docs/atlas/atlas-stream-processing/stream-aggregation-stages.md#std-label-atlas-sp-aggregation) the stream processor applies to your streaming data. |
| `options` | object | Optional | Object that defines optional settings for your stream processor. |
| `options.limit` | integer | Optional | Maximum number of documents to return to your terminal. `sp.process()` ends the session after it returns this number of documents. |
| `options.tier` | string | Optional | Tier to assign to the stream processor. Tier names are case-insensitive. If you omit this field, the stream processor uses the default tier for your stream processing workspace. Must be one of the following values: `SP2`; `SP5`; `SP10`; `SP30`; `SP50` To learn more, see [Tiers.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-atlas-sp-architecture-tiers) |

To create a stream processor interactively:

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)

   **Example:**

   The following command connects to a stream processing workspace as a user named `streamOwner` using [x.509](https://www.mongodb.com/docs/manual/core/security-x.509/) authentication:

   ```sh
   mongosh "mongodb://atlas-stream-78xq9gi1v4b4288x06a73e9f-zi30g.virginia-usa.a.query.mongodb-qa.net/?authSource=%24external&authMechanism=MONGODB-X509" \\
   --tls --authenticationDatabase admin --username streamOwner
   ```

   Provide your user password when prompted.

2. Define a pipeline.

   In the [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh) prompt, assign an array containing the aggregation stages you want to apply to a variable named `pipeline`.

   The following example uses the `stuff` topic in the `myKafka` connection in the connection registry as the [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source), matches records where the `temperature` field has a value of `46` and emits the processed messages to the `output` topic of the `mySink` connection in the connection registry:

   ```sh
   pipeline = [
    {$source: {"connectionName": "myKafka", "topic": "stuff"}},
    {$match: { temperature: 46 }},
    {
      "$emit": {
        "connectionName": "mySink",
        "topic" : "output",
      }
    }
   ]
   ```

3. Create a stream processor.

   The following command creates a stream processor that applies the logic defined in `pipeline`.

   ```sh
   sp.process(pipeline)
   ```

## Create a Stream Processor

The Atlas Administration API provides an endpoint for creating a stream processor.

[Create One Stream Processor](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-creategroupstreamprocessor)

To create a stream processor in the Atlas UI, go to the Stream Processing page for your Atlas project and click Manage in the pane for your stream processing workspace.

You can choose between using the Visual Builder or the JSON (Javascript Object Notation) editor to configure your stream processor:

### Visual Builder

1. Open the Visual Builder.

   Click Stream Processors in the left-hand navigation menu, then click Create stream
   processor.

   The Create Stream Processor screen opens with a form where you can configure your stream processor.

2. Add a source connection.

   In the Build Pipeline section, select the connection that your stream processor reads data from in the Source drop-down list.

   This opens a JSON (Javascript Object Notation) text box that contains the `$source` stage for your stream processor. In the text box, set `db` to the name of your database and `coll` to the name of the collection in that database. To learn more about `$source` stage syntax, see [`$source`.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source)

3. Optional: Add aggregation stages to the pipeline.

   In the Pipeline field, select the aggregation stages that you want your stream processor to apply. To add these stages later, leave this field empty.

   To learn more about stream processing aggregation stages and their syntax, 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)

   **Example:**

   The following [`$match`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/match.md#mongodb-pipeline-pipe.-match) stage matches all documents in the pre-configured `sample_stream_solar` stream where the `obs.watts` field is greater than `300`:

   ```json
   {
     "$match": {
       "obs.watts": { "$gt": 300 }
     }
   }
   ```

4. Add a sink connection.

   In the Sink field, select the connection that your stream processor writes processed data to.

   This opens a JSON (Javascript Object Notation) text box that contains the `$merge` stage for your stream processor. In the text box, set `db` to the name of your database and `coll` to the name of the collection in that database. To learn more about `$merge` stage syntax, see [`$merge`.](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge.md#mongodb-pipeline-pipe.-merge)

   **Example:**

   The following `$merge` stage writes processed data to the `demoDb.demoColl` collection in a connection named `demoConnection`:

   ```json
   {
     "$merge": {
       "into": {
         "connectionName": "demoConnection",
         "db": "demoDb",
         "coll": "demoColl"
       }
     }
   }
   ```

5. Enter your stream processor details.

   In the Enter processor details section, do the following:

   In the Stream processor name field, enter a name for your stream processor.

   Select the data processing tier for your stream processor. The default tier is `SP10`. To learn more, see [Tiers.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-atlas-sp-architecture-tiers)

   Specify whether to start your stream processor when Atlas Stream Processing creates it.

6. (Optional) Add a dead letter queue.

   Specify a destination for your [dead letter queue.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-atlas-sp-dlq)

7. (Optional) Create failover processors.

   In the Advanced settings pane, toggle Create failover processor(s) to create one or more [failover processors.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-atlas-sp-architecture-failover)

   IMPORTANT: This feature is only available in stream processing workspaces configured with failover regions. To learn more, see [Limitations.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/limitations.md#std-label-atlas-sp-limitations)

8. Click Create stream processor.

   Atlas Stream Processing creates the stream processor and lists it on the Stream Processors tab of the Stream Processing page.

To create one stream processor within the stream processing workspace you specify using the Atlas CLI, run the following command:

```sh

atlas api streams createStreamProcessor [options]

```

To learn more about the command syntax and parameters, see the Atlas CLI documentation for [atlas api streams createStreamProcessor.](https://www.mongodb.com/docs/atlas/cli/current/command/atlas-api-streams-createStreamProcessor/)

To create a new stream processor with [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh), use the [`sp.createStreamProcessor()`](https://www.mongodb.com/docs/manual/reference/method/sp.createStreamProcessor.md#mongodb-method-sp.createStreamProcessor) method. It has the following syntax:

```sh
sp.createStreamProcessor(<name>, <pipeline>, <options>)
```

| Argument | Type | Necessity | Description |
| --- | --- | --- | --- |
| `name` | string | Required | Logical name for the stream processor. This must be unique within the stream processing workspace. This name should contain only alphanumeric characters. |
| `pipeline` | array | Required | [Stream aggregation pipeline](https://www.mongodb.com/docs/atlas/atlas-stream-processing/stream-aggregation-stages.md#std-label-atlas-sp-aggregation) you want to apply to your streaming data. |
| `options` | object | Optional | Object defining various optional settings for your stream processor. |
| `options.dlq` | object | Conditional | Object assigning a [dead letter queue](https://www.mongodb.com/docs/manual/reference/glossary.md#std-term-dead-letter-queue) for your stream processing workspace. This field is necessary if you define the `options` field. |
| `options.dlq.connectionName` | string | Conditional | Human-readable label that identifies a connection in your connection registry. This connection must reference an Atlas cluster. This field is necessary if you define the `options.dlq` field. |
| `options.dlq.db` | string | Conditional | Name of an Atlas database on the cluster specified in `options.dlq.connectionName`. This field is necessary if you define the `options.dlq` field. |
| `options.dlq.coll` | string | Conditional | Name of a collection in the database specified in `options.dlq.db`. This field is necessary if you define the `options.dlq` field. |
| `options.tier` | string | Optional | The tier of the pod to which Atlas Stream Processing assigns the processor. If you do not declare this option, Atlas Stream Processing assigns the processor to a pod of the stream processing workspace's default tier. To learn more, see [Tiers.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-atlas-sp-architecture-tiers) |
| `options.failover` | boolean | Optional | Flag that specifies whether to create the processor in every [failover region](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-atlas-sp-architecture-failover) configured for your stream processing workspace in addition to the default region. If you omit this field, Atlas Stream Processing creates the processor only in the default region. |

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 run it to connect to your stream processing workspace.

   **Example:**

   The following command connects to a stream processing workspace as a user named `streamOwner` using [x.509](https://www.mongodb.com/docs/manual/core/security-x.509/) authentication.

   ```sh
   mongosh "mongodb://atlas-stream-78xq9gi1v4b4288x06a73e9f-zi30g.virginia-usa.a.query.mongodb-qa.net/?authSource=%24external&authMechanism=MONGODB-X509" \\
   --tls --authenticationDatabase admin --username streamOwner
   ```

   Provide your user password when prompted.

2. Define a pipeline.

   In the [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh) prompt, assign an array containing the aggregation stages you want to apply to a variable named `pipeline`.

   The following example pipeline uses the `stuff` topic in the  `myKafka` connection in the connection registry as the [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source), matches records where the `temperature` field has a value of `46` and emits the processed messages to the `output` topic of the `mySink` connection in the connection registry:

   ```sh
   pipeline = [
     {$source: {"connectionName": "myKafka", "topic": "stuff"}},
     {$match: { temperature: 46 }},
     {
       "$emit": {
         "connectionName": "mySink",
         "topic" : "output",
       }
     }
   ]
   ```

3. (Optional) Define a [DLQ](https://www.mongodb.com/docs/manual/reference/glossary.md#std-term-dead-letter-queue).

   In the [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh) prompt, assign an object containing the following properties of your DLQ:

   - Connection name

   - Database name

   - Collection name

   The following example defines a DLQ over the `cluster01` connection, in the `metadata.dlq` database collection.

   ```sh
   deadLetter = {
     dlq: {
       connectionName: "cluster01",
       db: "metadata",
       coll: "dlq"
     }
   }
   ```

4. Create a stream processor.

   The following command creates a stream processor named `proc01` that applies the logic defined in `pipeline`. Documents that throw errors in processing are written to the DLQ defined in `deadLetter`.

   ```sh
   sp.createStreamProcessor("proc01", pipeline, deadLetter)
   ```

## Start a Stream Processor

**Note:**

Atlas Stream Processing discards the internal state of stream processors which have been `stopped` for 45 days or more. When you start such a processor, it operates and reports [statistics](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-stream-processor.md#std-label-atlas-sp-manage-processor-stats) identically to its initial run.

The Atlas Administration API provides the [Start One Stream Processor](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-startgroupstreamprocessor) and [Start One Stream Processor with Options](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-startgroupstreamprocessorwith) endpoints for starting a stream processor.

- To start a stream processor without options, use [Start One Stream Processor](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-startgroupstreamprocessor)

- To start a stream processor and modify properties of the [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage, use [Start One Stream Processor with Options](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-startgroupstreamprocessorwith) This endpoint supports modifying either the `startAfter` or `startAtOperationTime` properties.

To start a stream processor in the Atlas UI, go to the Stream Processing page for your Atlas project and click Manage in the pane for your stream processing workspace to view the list of stream processors defined for it.

Then, click the Start icon for your stream processor.

To start one stream processor within the stream processing workspace you specify using the Atlas CLI, run the following command:

```sh

atlas api streams startStreamProcessor [options]

```

To learn more about the command syntax and parameters, see the Atlas CLI documentation for [atlas api streams startStreamProcessor.](https://www.mongodb.com/docs/atlas/cli/current/command/atlas-api-streams-startStreamProcessor/)

To start a stream processor with [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh), use the [`sp.processor.start()`](https://www.mongodb.com/docs/manual/reference/method/sp.processor.start.md#mongodb-method-sp.processor.start) method. It has the following syntax:

```sh
sp.processor.start(<options>)
```

Where `<options>` is a generic, optional document whose fields are passed to the underlying start command. To learn more about the command options, see [`sp.createStreamProcessor()`](https://www.mongodb.com/docs/manual/reference/method/sp.createStreamProcessor.md#mongodb-method-sp.createStreamProcessor). For example, to start a stream processor named `proc01`, run the following command:

```shell
sp.proc01.start()
```

**Output:**

```shell
{ "ok" : 1 }
```

This method returns `{ "ok": 1 }` if the stream processor exists and isn't currently running. If you invoke `sp.processor.start()` for a stream processor that is not `STOPPED`, [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh) returns an error.

## Stop a Stream Processor

**Note:**

Atlas Stream Processing discards the internal state of stream processors which have been `stopped` for 45 days or more. When you start such a processor, it operates and reports [statistics](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-stream-processor.md#std-label-atlas-sp-manage-processor-stats) identically to its initial run.

The Atlas Administration API provides an endpoint for stopping a stream processor.

[Stop One Stream Processor](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-stopgroupstreamprocessor)

To stop a stream processor in the Atlas UI, go to the Stream Processing page for your Atlas project and click Manage in the pane for your stream processing workspace to view the list of stream processors defined for it.

Then, click the Stop icon for your stream processor.

To stop one stream processor within the stream processing workspace you specify using the Atlas CLI, run the following command:

```sh

atlas api streams stopStreamProcessor [options]

```

To learn more about the command syntax and parameters, see the Atlas CLI documentation for [atlas api streams stopStreamProcessor.](https://www.mongodb.com/docs/atlas/cli/current/command/atlas-api-streams-stopStreamProcessor/)

To stop an existing stream processor with [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh), use the [`sp.processor.stop()`](https://www.mongodb.com/docs/manual/reference/method/sp.processor.stop.md#mongodb-method-sp.processor.stop) method. It has the following syntax:

```sh
sp.processor.stop(<options>)
```

Where `<options>` is a generic, optional document whose fields are passed to the underlying stop command.

For example, to stop a stream processor named `proc01`, run the following command:

```shell
sp.proc01.stop()
```

**Output:**

```shell
{ "ok" : 1 }
```

This method returns `{ "ok": 1 }` if the stream processor exists and is currently running. If you invoke `sp.processor.stop()` for a stream processor that is not `running`, [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh) returns an error.

## Modify a Stream Processor

You can modify the following elements of an existing stream processor:

- Name

- [Pipeline](https://www.mongodb.com/docs/atlas/atlas-stream-processing/stream-aggregation-stages.md#std-label-atlas-sp-aggregation)

- [Dead Letter Queue](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-atlas-sp-dlq)

To modify a stream processor, do the following:

The Atlas Administration API provides an endpoint for modifying a stream processor.

[Modify One Stream Processor](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-updategroupstreamprocessor)

1. Click Stream Processors in the left-hand navigation menu.

2. Select the stream processor that you want to modify.

   From the list of stream processors, click the one that you want to modify. The View Stream Processor screen opens.

3. Click Edit Processor.

   The stream processor fields become editable.

4. Make your changes to the stream processor.

   You can edit any of the fields available for your stream processor.

5. Click Save changes.

To modify one existing stream processor within the stream processing workspace you specify using the Atlas CLI, run the following command:

```sh

atlas api streams updateStreamProcessor [options]

```

To learn more about the command syntax and parameters, see the Atlas CLI documentation for [atlas api streams updateStreamProcessor.](https://www.mongodb.com/docs/atlas/cli/current/command/atlas-api-streams-updateStreamProcessor/)

Requires [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh) v2.3.4+.

Use the `sp.<streamprocessor>.modify()` command to modify an existing stream processor. `<streamprocessor>` must be the name of a stopped stream processor defined for the current stream processing workspace.

For example, to modify a stream processor named `proc01`, run the following command:

```javascript
 sp.proc1.modify(<pipeline>, {
    resumeFromCheckpoint: bool, // optional
    name: string, // optional
    dlq: string, // optional
    failover: bool, // optional
})
```

The `failover` option controls whether the processor runs in the [failover regions](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-atlas-sp-architecture-failover) configured for your stream processing workspace:

- Set `failover` to `true` to read the processor definition from the default region and create that processor in every failover region that doesn't already have it.

- Set `failover` to `false` to disable failover for a processor that has failover regions configured and revert it to running in a single region.

### Add a Stage to an Existing Pipeline

```javascript
 sp.createStreamProcessor("foo", [
   {$source: {
      connectionName: "StreamsAtlasConnection",
      db: "test",
      coll: "test"
  }},
  {$merge: {
      into: {
          connectionName: "StreamsAtlasConnection",
          db: "testout",
          coll: "testout"
      }
  }}
])
sp.foo.start();
```

```javascript
sp.foo.stop();
sp.foo.modify([
  {$source: {
    connectionName: "StreamsAtlasConnection",
    db: "test",
    coll: "test"
  }},
  {$match: {
    operationType: "insert"
  }},
  {$merge: {
     into: {
     connectionName: "StreamsAtlasConnection",
     db: "testout",
     coll: "testout2"
     }
  }}
]);
sp.foo.start();
```

### Modify the Input Source of a Stream Processor

```javascript
sp.foo.modify([
  {$source: {
    connectionName: "StreamsAtlasConnection",
    db: "test",
    coll: "test",
    config: {
      startAtOperationTime: new Date(now.getTime() - 5 * 60 * 1000)
    }
  }},
  {$match: {
    operationType: "insert"
  }},
  {$merge: {
    into: {
      connectionName: "StreamsAtlasConnection",
      db: "testout",
      coll: "testout2"
    }
  }}
], {resumeFromCheckpoint: false});
```

### Remove a Dead Letter Queue from a Stream Processor

```javascript
sp.foo.stop();
sp.foo.modify({dlq: {}})
sp.foo.start();
```

### Modify a Stream Processor with a Window

```javascript
sp.foo.stop();
sp.foo.modify([
  {$source: {
    connectionName: "StreamsAtlasConnection",
    db: "test",
    coll: "test"
  }},
  {$replaceRoot: {newRoot: "$fullDocument"}},
  {$match: {cost: {$gt: 500}}},
  {$tumblingWindow: {
    interval: {unit: "day", size: 1},
    pipeline: [
     {$group: {_id: "$customerId", sum: {$sum: "$cost"}, avg: {$avg: "$cost"}}}
    ]
  }},
  {$merge: {
    into: {
    connectionName: "StreamsAtlasConnection",
    db: "testout",
    coll: "testout"
   }
  }}
], {resumeFromCheckpoint: false});
sp.foo.start();
```

By default, modified processors restore from the last checkpoint. Alternatively, you can set `resumeFromCheckpoint=false`, in which case the processor only retains summary stats. When you modify a processor with open windows, the windows are entirely recomputed on the updated pipeline.

**Note:**

If you change the name of a stream processor for which you had [configured](https://www.mongodb.com/docs/atlas/configure-alerts.md#std-label-configure-alerts) the [Stream Processor State is failed](https://www.mongodb.com/docs/atlas/reference/alert-conditions.md#std-label-atlas-sp-alerts) alert by using an Operator (which contains matcher expressions like `is`, `contains`, and more), Atlas won't trigger alerts for the renamed stream processor if the matcher expression doesn't match the new name. To monitor the renamed stream processor, [reconfigure](https://www.mongodb.com/docs/atlas/configure-alerts.md#std-label-configure-alerts) the alert.

### Limitations

When the default setting `resumeFromCheckpoint=true` is enabled, the following limitations apply:

- You can't modify the `$source` stage.

- You can't modify the interval of your window.

- You can't remove a window.

- You can only modify a pipeline with a window if that window has either a `$group` or `$sort` stage in its inner pipeline.

- You can't change an existing window type. For example, you can't change from a `$tumblingWindow` to a `$hoppingWindow` or vice versa.

- Processors with windows may reprocess some data as a product of recalculating the windows.

- Per-operator statistics are not retained after a modify operation.

## Initiate Failover for One Stream Processor

To initiate failover of an individual stream processor, use the [Start One Stream Processor With Options](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-startgroupstreamprocessorwith) endpoint. In the `failover` field of the request body, provide the name of the cloud region in which the target failover processor lives. Initializing the failover processor will also deactivate the primary processor.

1. In Atlas, go to the Stream Processing page for your project.

   If it's not already displayed, select the organization that contains your project from the  Organizations menu in the navigation bar.

   If it's not already displayed, select your project from the Projects menu in the navigation bar.

   In the sidebar, click Stream Processing under the Streaming Data heading.

   The [Stream Processing](https://cloud.mongodb.com/go?l=https%3A%2F%2Fcloud.mongodb.com%2Fv2%2F%3Cproject%3E%23%2FstreamProcessing) page displays.

2. Initiate failover.

   Locate the overview panel of the stream processing workspace that hosts the target processor.

   Click the ellipsis Manage button.

   Click Stream Processors in the left-hand navigation menu.

   Locate the target processor from the list of processors in the workspace and click the arrow to expand that processor's region list. Locate the region to which you want to initiate failover and click the Failover button In the Actions column.

   In the modal window, select the [failover type](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-atlas-sp-architecture-failover). To perform a Forced failover, type `Force` in the confirmation prompt. For a Forced failover, select Clear checkpoint to skip resuming from the most recent checkpoint. By default, this checkbox is cleared.

   Click the Activate \[TYPE] failover button, where `[TYPE]` is your chosen failover type.

To start one stream processor with optional parameters set using the Atlas CLI, run the following command:

```sh

atlas api streams startStreamProcessorWith [options]

```

To learn more about the command syntax and parameters, see the Atlas CLI documentation for [atlas api streams startStreamProcessorWith.](https://www.mongodb.com/docs/atlas/cli/current/command/atlas-api-streams-startStreamProcessorWith/)

To initiate failover for an individual stream processor with [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh), use the [`sp.processor.start()`](https://www.mongodb.com/docs/manual/reference/method/sp.processor.start.md#mongodb-method-sp.processor.start) method and pass the `failover` option. Starting the failover processor also stops the processor in the currently active region.

The `failover` option has the following syntax:

```sh
sp.<streamprocessor>.start({
  failover: {
    region: "<region>",
    mode: "<mode>",
    dryRun: <boolean>
  }
})
```

| Field | Type | Necessity | Description |
| --- | --- | --- | --- |
| `failover` | object | Optional | Object defining the failover event you want to initiate. |
| `failover.region` | string | Required | Name of the cloud region in which the target failover processor lives. This region must be one of the failover regions configured for your stream processing workspace. To check which failover regions your stream processing workspace uses, see [View stream processing workspaces](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-processing-instance.md#std-label-atlas-sp-manage-spi-view). For the regions that Atlas Stream Processing supports, see [Atlas Stream Processing Regions.](https://www.mongodb.com/docs/atlas/atlas-stream-processing.md#std-label-atlas-sp-regions) |
| `failover.mode` | string | Optional | Type of failover event you want to initiate. Value must be either `GRACEFUL` or `FORCED`. To learn more, see [Failover Processors.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-atlas-sp-architecture-failover) |
| `failover.dryRun` | boolean | Optional | Flag that specifies whether to test the failover process without consuming data. Set this field to `true` to validate your failover configuration before you initiate a failover event in production. |

`GRACEFUL` mode takes a checkpoint and confirms that the active processor has stopped before it initiates failover. `FORCED` mode also attempts to stop the active processor, but initiates failover even if the processor doesn't stop. By default, a `FORCED` failover starts the processor in the target region using the most recent checkpoint.

To start the target processor without recovering from a checkpoint for a `FORCED` failover, set `clearCheckpoints` to `true`:

```sh
sp.<streamprocessor>.start({
  failover: {
    region: "<region>",
    mode: "FORCED"
  },
  clearCheckpoints: true
})
```

When you clear checkpoints, the target processor starts from the current position in the stream instead of recovering its most recent checkpoint. Clearing checkpoints might cause duplicate or missing records in the output.

For example, to initiate a graceful failover of a stream processor named `proc01` to the `us-west-2` region, run the following command:

```sh
sp.proc01.start({
  failover: {
    region: "us-west-2",
    mode: "GRACEFUL",
    dryRun: false
  }
})
```

To stop a stream processor, use the [`sp.processor.stop()`](https://www.mongodb.com/docs/manual/reference/method/sp.processor.stop.md#mongodb-method-sp.processor.stop) method. You don't have to stop the active processor before you initiate failover.

## Drop a Stream Processor

The Atlas Administration API provides an endpoint for deleting a stream processor.

[Delete One Stream Processor](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-deletegroupstreamprocessor)

1. Click Stream Processors in the left-hand navigation menu.

2. Select the stream processor that you want to delete.

   From the list of stream processors, click the one that you want to delete.

   If your stream processor has a status of `STARTED`, click the stop button in the Actions column to stop it first.

3. Click the ellipsis () button in the Actions column.

4. Select Delete stream processor from the menu.

   In the confirmation dialog that appears, type the name of the stream processor to confirm that you want to delete it, then click Delete.

To delete one stream processor within the stream processing workspace you specify using the Atlas CLI, run the following command:

```sh

atlas api streams deleteStreamProcessor [options]

```

To learn more about the command syntax and parameters, see the Atlas CLI documentation for [atlas api streams deleteStreamProcessor.](https://www.mongodb.com/docs/atlas/cli/current/command/atlas-api-streams-deleteStreamProcessor/)

To delete an existing stream processor with [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh), use the [`sp.processor.drop()`](https://www.mongodb.com/docs/manual/reference/method/sp.processor.drop.md#mongodb-method-sp.processor.drop) method. It has the following syntax:

```sh
sp.processor.drop(<options>)
```

Where `<options>` is a generic, optional document whose fields are passed to the underlying drop command.

For example, to drop a stream processor named `proc01`, run the following command:

```sh
sp.proc01.drop()
```

This method returns:

- `true` if the stream processor exists.

- `false` if the stream processor doesn't exist.

When you drop a stream processor, all resources that Atlas Stream Processing provisioned for it are destroyed, along with all saved state.

## List Available Stream Processors

The Atlas Administration API provides an endpoint for listing all available stream processors.

[List Stream Processors](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-getgroupstreamprocessors)

To view the list of stream processors defined for your stream processing workspace in the Atlas UI, go to the Stream Processing page for your Atlas project and click Manage in the pane for your stream processing workspace.

The list of stream processors and their statuses displays.

To return all stream processors within the stream processing workspace you specify using the Atlas CLI, run the following command:

```sh

atlas api streams getStreamProcessors [options]

```

To learn more about the command syntax and parameters, see the Atlas CLI documentation for [atlas api streams getStreamProcessors.](https://www.mongodb.com/docs/atlas/cli/current/command/atlas-api-streams-getStreamProcessors/)

To list all available stream processors on the current stream processing workspace with [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh), use the [`sp.listStreamProcessors()`](https://www.mongodb.com/docs/manual/reference/method/sp.listStreamProcessors.md#mongodb-method-sp.listStreamProcessors) method. This returns a list of documents containing the name, start time, current state, and pipeline associated with each stream processor. It has the following syntax:

```sh
sp.listStreamProcessors(<filter>)
```

`<filter>` is a document specifying which field(s) to filter the list by.

**Example:**

The following example shows a return value for an unfiltered request:

```sh
sp.listStreamProcessors()
```

**Output:**

```json
{
  id: '0135',
  name: "proc01",
  last_modified: ISODate("2023-03-20T20:15:54.601Z"),
  state: "RUNNING",
  error_msg: '',
  pipeline: [
    {
      $source: {
        connectionName: "myKafka",
        topic: "stuff"
      }
    },
    {
      $match: {
        temperature: 46
      }
    },
    {
      $emit: {
        connectionName: "mySink",
        topic: "output",
      }
    }
  ],
  lastStateChange: ISODate("2023-03-20T20:15:59.442Z")
},
{
  id: '0218',
  name: "proc02",
  last_modified: ISODate("2023-03-21T20:17:33.601Z"),
  state: "STOPPED",
  error_msg: '',
  pipeline: [
    {
      $source: {
        connectionName: "myKafka",
        topic: "things"
      }
    },
    {
      $match: {
        temperature: 41
      }
    },
    {
      $emit: {
        connectionName: "mySink",
        topic: "results",
      }
    }
  ],
  lastStateChange: ISODate("2023-03-21T20:18:26.139Z")
}
```

If you run the command again on the same stream processing workspace, filtering for a `"state"` of `"running"`, you see the following output:

```sh
sp.listStreamProcessors({"state": "running"})
```

**Output:**

```json
{
  id: '0135',
  name: "proc01",
  last_modified: ISODate("2023-03-20T20:15:54.601Z"),
  state: "RUNNING",
  error_msg: '',
  pipeline: [
    {
      $source: {
        connectionName: "myKafka",
        topic: "stuff"
      }
    },
    {
      $match: {
        temperature: 46
      }
    },
    {
      $emit: {
        connectionName: "mySink",
        topic: "output",
      }
    }
  ],
  lastStateChange: ISODate("2023-03-20T20:15:59.442Z")
}
```

## List Workspace Defaults

The [List Stream Workspace Details](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-getgroupstreamworkspace) endpoint lists the details of a stream processing workspace, including the `defaultTierSize` and `maxTierSize`.

To view the `defaultTierSize` and `maxTierSize` of a stream processing workspace in the Atlas UI, go to the Stream Processing page for your Atlas project.

The `defaultTierSize` and `maxTierSize` are listed at the bottom of the stream processing workspace card.

To return details of one stream processing workspace you specify using the Atlas CLI, run the following command:

```sh

atlas api streams getStreamWorkspace [options]

```

To learn more about the command syntax and parameters, see the Atlas CLI documentation for [atlas api streams getStreamWorkspace.](https://www.mongodb.com/docs/atlas/cli/current/command/atlas-api-streams-getStreamWorkspace/)

To view the `defaultTierSize` and `maxTierSize` of a stream processing workspace using [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh), use the [`sp.listWorkspaceDefaults()`](https://www.mongodb.com/docs/manual/reference/method/sp.listWorkspaceDefaults.md#mongodb-method-sp.listWorkspaceDefaults) method.

```sh
sp.listWorkspaceDefaults()
```

This method returns:

- `defaultTierSize`

- `maxTierSize`

## Sample from a Stream Processor

To return an array of sampled results from an existing stream processor to `STDOUT` with [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh), use the [`sp.processor.sample()`](https://www.mongodb.com/docs/manual/reference/method/sp.processor.sample.md#mongodb-method-sp.processor.sample) method. For example, the following command samples from a stream processor named `proc01`.

```sh
sp.proc01.sample()
```

This command runs continuously until you cancel it by using `CTRL-C`, or until the returned samples cumulatively reach 40 MB in size. The stream processor reports invalid documents in the sample in a `_dlqMessage` document of the following form:

```json&#xA;
 {
   _dlqMessage: {
     errInfo: {
    reason: "<reasonForError>"
     },
     doc: {
    _id: ObjectId('<group-id>'),
    ...
     },
     processorName: '<procName>',
     workspaceName: '<workspaceName>',
     dlqTime: ISODate('2024-09-19T20:04:34.263+00:00')
   }
}
```

You can use these messages to diagnose data hygiene issues without defining a [dead letter queue](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-atlas-sp-dlq) collection.

## View Statistics of a Stream Processor

**Note:**

Atlas Stream Processing discards the internal state of stream processors which have been `stopped` for 45 days or more. When you start such a processor, it operates and reports [statistics](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-stream-processor.md#std-label-atlas-sp-manage-processor-stats) identically to its initial run.

For a description of the metrics that Atlas Stream Processing reports for a stream processor, see [Stream Processor Metrics.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/monitoring.md#std-label-atlas-sp-metrics)

The Atlas Administration API provides an endpoint for viewing the statistics of a stream processor.

[Get One Stream Processor](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-getgroupstreamprocessor)

To view monitoring for your stream processor, go to the Stream Processing page for your Atlas project and open the Monitoring tab. Then, select your stream processor from the Stream processor drop-down list at the top left of the page.

To return a document summarizing the current status of an existing stream processor with [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh), use the [`sp.processor.stats()`](https://www.mongodb.com/docs/manual/reference/method/sp.processor.stats.md#mongodb-method-sp.processor.stats) method. It has the following syntax:

```sh
sp.<streamprocessor>.stats({options: {<options>}})
```

Where `options` is an optional document. To learn more about the command options and output fields, see [`sp.processor.stats()`.](https://www.mongodb.com/docs/manual/reference/method/sp.processor.stats.md#mongodb-method-sp.processor.stats)

For a description of every field in the output document, see [Stream Processor Metrics.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/monitoring.md#std-label-atlas-sp-metrics)

For example, the following shows the status of a stream processor named `proc01` on a stream processing workspace named `inst01` with item sizes displayed in KB:

```sh
sp.proc01.stats(1024)

{
  ok: 1,
  ns: 'inst01',
  stats: {
    name: 'proc01',
    status: 'running',
    scaleFactor: Long("1"),
    inputMessageCount: Long("706028"),
    inputMessageSize: 958685236,
    outputMessageCount: Long("46322"),
    outputMessageSize: 85666332,
    dlqMessageCount: Long("0"),
    dlqMessageSize: Long("0"),
    stateSize: Long("2747968"),
    watermark: ISODate("2023-12-14T14:35:32.417Z"),
    addedParallelism: 7,
    ok: 1
  },
}
```

The example shows the overall statistics for the stream processor.

To see how individual operators behave or how much traffic each target handles, call `sp.<streamprocessor>.stats()` with the `verbose` option and inspect `stats.operatorStats` and, for some operators, `stats.operatorStats.targetStats`.

For example, for a source operator, `stats.operatorStats.targetStats` collects the `inputMessageCount` and `inputMessageSize` fields for every unique `db`/`coll` or every unique topic:

```sh
{
  "name" : "KafkaConsumerOperator",
  "inputMessageCount" : NumberLong(100),
  "inputMessageSize" : 100352,
  "targetStats" : [
    {
    "topic" : "outputTopic1",
    "inputMessageCount" : NumberLong(100),
    "inputMessageSize" : 100352
    }
  ],
  ...
}
```

And for a sink operator, `stats.operatorStats.targetStats` collects the `outputMessageCount` and `outputMessageSize` fields for every unique `db`/`coll` or every unique topic:

```sh
{
  "name" : "MergeOperator",
  "inputMessageCount" : NumberLong(10),
  "inputMessageSize" : 1744,
  "targetStats" : [
    {
      "db" : "cust1",
      "coll" : "outColl1",
      "outputMessageCount" : NumberLong(3),
      "outputMessageSize" : 1748
    },
    {
      "db" : "cust2",
      "coll" : "outColl2",
      "outputMessageCount" : NumberLong(4),
      "outputMessageSize" : 2241
    }
  ],
  ...
}
```