Supported Aggregation Pipeline Stages

On this page

Supported Aggregation Pipeline Stages

Atlas Stream Processing extends the aggregation pipeline with stages for processing continuous data streams. These stages combine with existing aggregation stages built in to the default 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.

Warning

The stream_meta field can only be removed with a $project or an $unset stage. You can only remove this field with one of the two explicit calls as defined below:

{$project : { _stream_meta : 0 }}

{$unset : ["_stream_meta"]}

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`	Specifies a streaming data source to consume messages from.
`$validate`	Validates the documents of a stream against a user-defined schema.
$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` stage requires that you specify a Atlas collection in the Connection Registry as the value for the `from` field.
`$hoppingWindow`	Assigns documents from a stream to windows with user-defined durations and intervals between start times.
`$tumblingWindow`	Assigns documents from a stream to non-overlapping, continuous windows with user-defined durations.
$emit	Specifies a stream or time series collection in the connection registry to emit messages to.
$merge	A version of the existing `$merge` stage where the value of the `connectionName` field must always be the name of a remote collection in the Connection Registry.

You can also use the following stages supported by all mongod processes in your streaming data pipelines:

Aggregation Pipeline Stage	Use Conditions
`$addFields`	Anywhere
`$match`	Anywhere
`$project`	Anywhere
`$redact`	Anywhere
`$replaceRoot`	Anywhere
`$replaceWith`	Anywhere
`$set`	Anywhere
`$unset`	Anywhere
`$unwind`	Anywhere
`$group`	Only within `$hoppingWindow` or `$tumblingWindow` stages.
`$sort`	Only within `$hoppingWindow` or `$tumblingWindow` stages.
`$limit`	Only within `$hoppingWindow` or `$tumblingWindow` stages.
`$count`	Only within `$hoppingWindow` or `$tumblingWindow` stages.

Supported Aggregation Pipeline Stages

Atlas Stream Processing provides a number of extensions to the core MongoDB Aggregation Pipeline syntax. To learn more about these extensions, see Supported Aggregation Pipeline Stages.

Certain core Aggregation Pipeline stages have limited support, or are unsupported as noted in the following table. If a stage is not listed in the table, Atlas Stream Processing supports it.

Aggregation Stage	Support Status
`$group`	Only supported in a `$tumblingWindow` or `$hoppingWindow`.
`$sort`	Only supported in a `$tumblingWindow` or `$hoppingWindow`.
`$limit`	Only supported in a `$tumblingWindow` or `$hoppingWindow`.
`$count`	Only supported in a `$tumblingWindow` or `$hoppingWindow`.
`$merge`	Supported with modified syntax. For more information, see `$merge`.
`$lookup`	Supported with modified syntax. For more information, see `$lookup`.

Back

$convert

$source