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

# Aggregation Pipeline Stages for Stream Processing

Atlas Stream Processing extends the [aggregation pipeline](https://www.mongodb.com/docs/manual/aggregation/) with stages for processing continuous data streams. These stages combine with existing aggregation stages built in to the default [`mongod`](https://www.mongodb.com/docs/manual/reference/program/mongod.md#mongodb-binary-bin.mongod) process, enabling you to perform many of the same operations on continuous data as you can perform on data-at-rest. An Atlas Stream Processing pipeline definition cannot exceed 16 MB.

The following table lists the aggregation pipeline stages unique to Atlas Stream Processing and those which exist in modified form in Atlas Stream Processing.

**Note:**

For details on a specific operator, including syntax and examples, click on the link to the operator's reference page.

| Aggregation Pipeline Stage | Purpose |
| --- | --- |
| [$source](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#std-label-atlas-sp-agg-source) | Specifies a streaming data source to consume messages from. |
| [$validate](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-validate.md#std-label-streams-agg-pipeline-validate) | Validates the documents of a stream against a user-defined schema. |
| [$https](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-https.md#std-label-atlas-sp-agg-https) | Supports `curl`-like calls to arbitrary REST API endpoints within your stream processor. |
| [$lookup](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-lookup.md#std-label-streams-agg-pipeline-lookup) | Performs a left outer join to a specified collection to filter in documents from the "joined" collection for processing. This version of the existing [`$lookup`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/lookup.md#mongodb-pipeline-pipe.-lookup) stage requires that you specify a Atlas collection in the [Connection Registry](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-connection-registry.md#std-label-atlas-sp-manage-connections) as the value for the `from` field. |
| [$hoppingWindow](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-hopping.md#std-label-streams-agg-pipeline-hopping) | Assigns documents from a stream to [windows](https://www.mongodb.com/docs/atlas/atlas-stream-processing/windows.md#std-label-atlas-sp-windows) with user-defined durations and intervals between start times. |
| [$tumblingWindow](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-tumbling.md#std-label-streams-agg-pipeline-tumbling) | Assigns documents from a stream to non-overlapping, continuous [windows](https://www.mongodb.com/docs/atlas/atlas-stream-processing/windows.md#std-label-atlas-sp-windows) with user-defined durations. |
| [$sessionWindow](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-session.md#std-label-streams-agg-pipeline-session) | Runs a pipeline on each "session" of activity in an input stream. Each document is assigned to a session window based on its partition and timestamp. |
| [$setStreamMeta](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-setstreammeta.md#std-label-atlas-sp-agg-setStreamMeta) | Sets a metadata field on documents from an input stream. This metadata field can be used to perform dynamic operations such as sink routing on documents without changing their body content. |
| [$cachedLookup](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-cachedlookup.md#std-label-atlas-sp-agg-cachedlookup) | Performs a lookup with a time-to-live cache to reduce repeated reads from the joined collection. |
| [$externalFunction](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-externalfunction.md#std-label-atlas-sp-agg-external-function) | Calls an AWS (Amazon Web Services) Lambda function within the pipeline, optionally passing the result to downstream stages. |
| [$throttle](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-throttle.md#std-label-atlas-sp-agg-throttle) | Limits the rate at which the stream processor passes data to downstream stages, in bytes or messages per second. |
| [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-streams-agg-pipeline-emit) | Specifies a stream or time series collection in the connection registry to emit messages to. |
| [$merge](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-merge.md#std-label-streams-agg-pipeline-merge) | A version of the existing [`$merge` Stage (Atlas Data Federation)](https://www.mongodb.com/docs/atlas/data-federation/supported-unsupported/pipeline/merge.md#std-label-adf-merge-stage) stage where the value of the `connectionName` field must always be the name of a remote collection in the [Connection Registry.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-connection-registry.md#std-label-atlas-sp-manage-connections) |
| [$vectorSearch](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-vectorsearch.md#std-label-atlas-sp-agg-vectorSearch) | Queries a MongoDB Vector Search index on a collection in the [Connection Registry](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-connection-registry.md#std-label-atlas-sp-manage-connections) and attaches the results to each streaming document. |

## Supported Aggregation Pipeline Stages

Atlas Stream Processing provides several extensions to the core MongoDB [Aggregation Pipeline](https://www.mongodb.com/docs/manual/core/aggregation-pipeline/) syntax. To learn more about these extensions, see [Supported Aggregation Pipeline Stages.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/stream-aggregation-stages.md#std-label-atlas-sp-aggregation)

Certain core aggregation pipeline stages are only supported when nested within other pipeline stages. The following table shows which aggregation pipeline stages are supported for use within the main pipeline or when nested within the specified aggregation pipeline stages.

| Aggregation Stage | Supported in Main Pipeline | Supported in [`$tumblingWindow`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-tumbling.md#mongodb-pipeline-pipe.-tumblingWindow), [`$hoppingWindow`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-hopping.md#mongodb-pipeline-pipe.-hoppingWindow), and [`$sessionWindow`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-session.md#mongodb-pipeline-pipe.-sessionWindow) | Supported in [`$https`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-https.md#mongodb-pipeline-pipe.-https) and [`$externalFunction`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-externalfunction.md#mongodb-pipeline-pipe.-externalFunction) |
| --- | --- | --- | --- |
| [`$addFields`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/addFields.md#mongodb-pipeline-pipe.-addFields) | √ | √ | √ |
| [`$project`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/project.md#mongodb-pipeline-pipe.-project) | √ | √ | √ |
| [`$replaceRoot`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/replaceRoot.md#mongodb-pipeline-pipe.-replaceRoot) | √ | √ | √ |
| [`$set`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/set.md#mongodb-pipeline-pipe.-set) | √ | √ | √ |
| [`$redact`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/redact.md#mongodb-pipeline-pipe.-redact) | √ | √ | |
| [`$match`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/match.md#mongodb-pipeline-pipe.-match) | √ | √ | |
| [`$replaceWith`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/replaceWith.md#mongodb-pipeline-pipe.-replaceWith) | √ | √ | |
| [`$unset`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/unset.md#mongodb-pipeline-pipe.-unset) | √ | √ | |
| [`$unwind`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/unwind.md#mongodb-pipeline-pipe.-unwind) | √ | √ | |
| [`$validate`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-validate.md#mongodb-pipeline-pipe.-validate) | √ | √ | |
| [`$https`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-https.md#mongodb-pipeline-pipe.-https) | √ | √ | |
| [`$externalFunction`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-externalfunction.md#mongodb-pipeline-pipe.-externalFunction) | √ | √ | |
| [`$lookup`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/lookup.md#mongodb-pipeline-pipe.-lookup) | √ | √ | |
| [`$merge`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge.md#mongodb-pipeline-pipe.-merge) | √ |  | |
| [`$tumblingWindow`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-tumbling.md#mongodb-pipeline-pipe.-tumblingWindow) | √ |  | |
| [`$hoppingWindow`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-hopping.md#mongodb-pipeline-pipe.-hoppingWindow) | √ |  | |
| [`$sessionWindow`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-session.md#mongodb-pipeline-pipe.-sessionWindow) | √ |  | |
| [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit) | √ |  | |
| [$cachedLookup](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-cachedlookup.md#std-label-atlas-sp-agg-cachedlookup) | √ |  | |
| [$setStreamMeta](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-setstreammeta.md#std-label-atlas-sp-agg-setStreamMeta) | √ |  | |
| [$throttle](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-throttle.md#std-label-atlas-sp-agg-throttle) | √ |  | |
| [`$group`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/group.md#mongodb-pipeline-pipe.-group) |  | √ | |
| [`$sort`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/sort.md#mongodb-pipeline-pipe.-sort) |  | √ | |
| [`$count`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/count.md#mongodb-pipeline-pipe.-count) |  | √ | |
| [`$limit`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/limit.md#mongodb-pipeline-pipe.-limit) |  | √ | |

Supported with modified syntax.