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

# `$source` Stage (Stream Processing)

## Definition

The [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage specifies a connection in the [Connection Registry](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-connection-registry.md#std-label-atlas-sp-manage-connections) to stream data from. The following connection types are supported:

- [Apache Kafka](https://kafka.apache.org/) broker

- MongoDB collection change stream

- MongoDB database change stream

- MongoDB cluster change stream

- AWS (Amazon Web Services) Kinesis data stream

- Document array

- Cron schedule

## Syntax

### Apache Kafka Broker

To operate on streaming data from an [Apache Kafka](https://kafka.apache.org/) broker, the `$source` stage has the following prototype form:

```json
{
  "$source": {
    "connectionName": "<registered-connection>",
    "topic" : ["<source-topic>", ...],
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "partitionIdleTimeout": {
      "size": <duration-number>,
      "unit": "<duration-unit>"
    },
    "partitions": {
      "min": <partition-id>,
      "max": <partition-id>
    },
    "schemaRegistry": {
      "connectionName": "<schema-registry-name>",
    },
    "config": {
      "auto_offset_reset": "<start-event>",
      "group_id": "<group-id>",
      "keyFormat": "<deserialization-type>",
      "keyFormatError": "<error-handling>"
    },
  }
}
```

The `$source` stage takes a document with the following fields:

| Field | Type | Necessity | Description |
| --- | --- | --- | --- |
| `connectionName` | string | Required | Label that identifies the connection in the [Connection Registry](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-connection-registry.md#std-label-atlas-sp-manage-connections), to ingest data from. |
| `topic` | string or array of strings | Required | Name of one or more [Apache Kafka](https://kafka.apache.org/) topics from which to stream messages. If you want to stream messages from more than one topic, specify them in an array. |
| `timeField` | document | Optional | Document that defines an authoritative timestamp for incoming messages. If you use `timeField`, you must define it as one of the following: a :expression:`$toDate` that takes a source message field as an argument.; a :expression:`$dateFromString` that takes a source message field as an argument. If you do not declare a `timeField`, Atlas Stream Processing creates a timestamp from the message timestamp provided by the source. |
| `partitionIdleTimeout` | document | Optional | Document specifying the amount of time that a partition is allowed to be idle before it is ignored in watermark calculations. This field is disabled by default. To handle partitions that don't move forward due to idleness, set a value for this field. |
| `partitionIdleTimeout.size` | integer | Optional | Number specifying the duration of the partition idle timeout. |
| `partitionIdleTimeout.unit` | string | Optional | Unit of time for the duration of the partition idle timeout. The value of `unit` can be one of the following: `"ms"` (millisecond); `"second"`; `"minute"`; `"hour"`; `"day"` |
| `partitions` | document | Optional | Document that specifies an inclusive range of partition IDs that the stream processor reads from. Atlas Stream Processing reads only from partitions of the source topic whose IDs fall within this range. If you specify more than one topic, Atlas Stream Processing applies the range to the partitions of each topic. If you omit this field, Atlas Stream Processing reads from all partitions of the source topic. |
| `partitions.min` | integer | Conditional | Lowest partition ID in the range. Must be `0` or greater. Required if you specify `partitions`. |
| `partitions.max` | integer | Optional | Highest partition ID in the range. Must be greater than or equal to `partitions.min`. If you omit this field, Atlas Stream Processing reads from all partitions with an ID greater than or equal to `partitions.min`, including partitions added to the topic after the stream processor starts. |
| `schemaRegistry` | document | Optional | Document enabling the use of a [Schema Registry](https://docs.confluent.io/platform/current/schema-registry/index.html) to support reading from an [Avro-serialized](https://avro.apache.org/docs/) source. To enable this feature, you must create a [Schema Registry connection.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/add-sp-connection.md#std-label-atlas-sp-add-connection) |
| `schemaRegistry.connectionName` | string | Conditional | Name of the Schema Registry connection to use for Avro deserialization. |
| `config` | document | Optional | Document containing fields that override various default values. |
| `config.auto_offset_reset` | string | Optional | Specifies which event in the [Apache Kafka](https://kafka.apache.org/) source topic to begin ingestion with. `auto_offset_reset` takes the following values: `end`, `latest`, or `largest` : to begin ingestion from the latest event in the topic at the time the aggregation is initialized.; `earliest`, `beginning`, or `smallest` : to begin ingestion from the earliest event in the topic. Defaults to `latest`. |
| `config.group_id` | string | Optional | ID of the kafka consumer group to associate with the stream processor. If omitted, Atlas Stream Processing associates the stream processing workspace with an auto-generated ID in the following format: `asp-${streamProcessorId}-consumer` Atlas Stream Processing automatically generates a value for this parameter for all persistent stream processors. For ephemeral stream processors defined with [sp.process()](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-stream-processor.md#std-label-atlas-sp-manage-processor-interactive), this parameter is set only if you manually define it. |
| `config.enable_auto_commit` | boolean | Conditional | Flag that determines the commit policy for Kafka broker partition offsets. Atlas Stream Processing supports two commit policies: If you set this parameter to `true`, Atlas Stream Processing commits offsets each time the `$source` stage passes data to the next operator.; If you set this parameter to `false`, stream processors commit partition offsets when Atlas Stream Processing takes a checkpoint. For an ephemeral stream processor defined with [sp.process()](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-stream-processor.md#std-label-atlas-sp-manage-processor-interactive), this parameter defaults to `false` unless you set `group_id`. Otherwise defaults to `true`. For more information on offsets when using Kafka as a `$source`, see [Kafka Sources and Consumer Group Offsets.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-kafka-source-offsets) |
| `config.keyFormat` | string | Optional | Data type used to deserialize [Apache Kafka](https://kafka.apache.org/) key data. Must be one of the following values: `"binData"`; `"string"`; `"json"`; `"int"`; `"long"` Defaults to `binData`. |
| `config.keyFormatError` | string | Optional | How to handle errors encountered when deserializing [Apache Kafka](https://kafka.apache.org/) key data. Must be one of the following values: `dlq`, which writes the document to your [Dead Letter Queue.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-atlas-sp-dlq); `passThrough`, which sends the document to the next stage without key data. |

**Note:**

Atlas Stream Processing requires that documents in the source data stream be valid `json` or `ejson`. Atlas Stream Processing sets the documents that don't meet this requirement to your [dead letter queue](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-atlas-sp-dlq) if you have configured one.

### MongoDB Collection Change Stream

An Atlas collection change stream allows applications to access real-time data changes on a single collection. To learn how to open a change stream against a collection, see [Change Streams.](https://www.mongodb.com/docs/manual/changeStreams/)

When using a change stream `$source`, configure the source cluster with an [oplog window](https://www.mongodb.com/docs/atlas/cluster-additional-settings.md#std-label-set-oplog-min-window) of at least 24 hours.

To read the change stream, Atlas Stream Processing scans the oplog collection. You may see `COLLSCAN` warnings in your logs as a result. These warnings indicate normal behavior and do not signal an error.

If you configure `config.fullDocument` or `config.fullDocumentBeforeChange` to `required`, enable `changeStreamPreAndPostImages` on each collection before any write operations that you want to capture. If a post-image is unavailable for an event because you had not enabled the feature when the write occurred, or because the post-image expired, the stream processor fails. To learn how to enable pre- and post-images, see [Change Streams with Document Pre- and Post-Images.](https://www.mongodb.com/docs/manual/reference/command/collMod.md#std-label-collMod-change-stream-pre-and-post-images)

**Important:**

You can [resume a change stream](https://www.mongodb.com/docs/manual/changeStreams.md#std-label-change-stream-resume) only while the oplog collection still holds the operation that its [resume token](https://www.mongodb.com/docs/manual/changeStreams.md#std-label-change-stream-resume-token) identifies. To ensure that your applications can resume their change streams, configure a [Minimum Oplog Window](https://www.mongodb.com/docs/atlas/cluster-additional-settings.md#std-label-set-oplog-min-window) longer than the longest interruption you expect.

To operate on streaming data from an Atlas collection change stream, the `$source` stage has the following prototype form:

```json
{
  "$source": {
    "connectionName": "<registered-connection>",
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "db" : "<source-db>",
    "coll" : ["<source-coll>",...],
    "initialSync": {
      "enable": <boolean>,
      "parallelism": <integer>
    },
    "readPreference": "<read-preference>",
    "readPreferenceTags": [
      {"<key>": "<value>"},
      . . .
    ],
    "config": {
      "startAfter": <start-token> | "startAtOperationTime": <timestamp>,
      "fullDocument": "<full-doc-condition>",
      "fullDocumentOnly": <boolean>,
      "fullDocumentBeforeChange": "<before-change-condition>",
      "pipeline": [{
        "<aggregation-stage>" : {
          <stage-input>,
          . . .
        },
        . . .
      }],
      "maxAwaitTimeMS": <time-ms>,
    }
  }
}
```

The `$source` stage takes a document with the following fields:

| Field | Type | Necessity | Description |
| --- | --- | --- | --- |
| `connectionName` | string | Conditional | Label that identifies the connection in the [Connection Registry](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-connection-registry.md#std-label-atlas-sp-manage-connections), to ingest data from. |
| `timeField` | document | Optional | Document that defines an authoritative timestamp for incoming messages. If you use `timeField`, you must define it as one of the following: a `$toDate` expression that takes a source message field as an argument; a `$dateFromString` expression that takes a source message field as an argument. If you do not declare a `timeField`, Atlas Stream Processing creates a timestamp from the message timestamp provided by the source. |
| `db` | string | Required | Name of a MongoDB database hosted on the Atlas instance specified by `connectionName`. The change stream of this database acts as the streaming data source. |
| `coll` | string or array of strings | Required | Name of one or more MongoDB collections hosted on the Atlas instance specified by `connectionName`. The change stream of these collections act as the streaming data source. If you omit this field, your stream processor sources from a [MongoDB Database Change Stream.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#std-label-atlas-sp-agg-source-syntax-db) |
| `initialSync` | document | Optional | Document containing `initialSync` configuration parameters. Atlas Stream Processing `initialSync` allows you to ingest preexisting documents in an Atlas collection as though they were insert `changeEvent` documents. If you enable `initialSync`, when you start your stream processor, it first ingests and processes all existing documents in the collection before proceeding to ingest and process new incoming `changeEvent` documents. Once the `initialSync` is complete, it doesn't repeat. If `coll` names more than one collection, Atlas Stream Processing synchronizes every collection in the list. To learn about multi-collection sync order, tier limits, and examples, see [Configure Multi-Collection Initial Sync.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/multi-collection-initial-sync.md#std-label-atlas-sp-multi-collection-initial-sync) Atlas Stream Processing fixes the set of collections to synchronize when the stream processor starts. It doesn't synchronize collections that you create after that point, and if synchronizing any one collection fails, the entire stream processor fails. If you enable `initialSync`, you can't use `$hoppingWindow`, `$sessionWindow`, or `$tumblingWindow` stages in your pipeline. IMPORTANT: See [Limitations](https://www.mongodb.com/docs/atlas/atlas-stream-processing/limitations.md#std-label-atlas-sp-initial-sync-limitations) for `initialSync` considerations and constraints. |
| `initialSync.enable` | boolean | Conditional | Determines whether to enable `initialSync`. If you declare an `initialSync` field, you must set this field. |
| `initialSync.parallelism` | integer | Optional | Determines the level of parallelism with which to process the `initialSync` operation. If you do not specify a value, it defaults to `1`. If `coll` names more than one collection, this value applies to the collection list as a whole rather than to each collection. `initialSync` applies this value to collections with any supported `_id` type, including types that Atlas Stream Processing can't read in a predictable order. The maximum value you can set depends on the tier of your stream processor. To learn more, 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) Each stream processor has a maximum cumulative parallelism value determined by its tier. The cumulative parallelism of a stream processor is calculated as follows: `parallelism total - parallelized stages` Where `parallelism total` is the sum of all `parallelism` values greater than 1 across the [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source), [$lookup](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-lookup.md#std-label-atlas-sp-agg-lookup), [$merge](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-merge.md#std-label-atlas-sp-agg-merge), [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit), and [$externalFunction](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-externalfunction.md#std-label-atlas-sp-agg-external-function) stages, and `parallelized stages` is the number of these stages with `parallelism` values greater than `1`. For example, if your `$source` stage sets a `parallelism` value of `4`, your `$lookup` stage sets no `parallelism` value (thus defaulting to `1`), and your `$merge` stage sets a `parallelism` value of `2`, then you have two `parallelized stages`, and the cumulative parallelism of your stream processor is calculated as `(4 + 2) - 2`. If a stream processor exceeds the maximum cumulative parallelism for its tier, Atlas Stream Processing throws an error and advises you of the minimum processor tier required for your intended level of parallelism. You must either scale the processor up to a higher tier or lower the parallelism values of your stages to resolve the error. To learn more, see [Stream Processing.](https://www.mongodb.com/docs/atlas/billing/stream-processing-costs.md#std-label-stream-processing-costs) |
| `readPreference` | string | Optional | [Read preference](https://www.mongodb.com/docs/manual/core/read-preference/) for change stream and `initialSync` operations. Defaults to `primary`. |
| `readPreferenceTags` | array | Optional | [Read preference tags](https://www.mongodb.com/docs/manual/core/read-preference-tags/) for change stream and `initialSync` operations. |
| `config` | document | Optional | Document containing fields that override various default values. |
| `config.startAfter` | token | Conditional | The change event after which the source begins reporting. This takes the form of a [resume token.](https://www.mongodb.com/docs/manual/changeStreams/#resume-tokens-from-change-events) You can use only one of either `config.startAfter` or `config.startAtOperationTime`. |
| `config.startAtOperationTime` | timestamp \| date | Conditional | The operation time after which the source should begin reporting. You can use only one of either `config.startAfter` or `config.startAtOperationTime`. Accepts [MongoDB Extended JSON](https://www.mongodb.com/docs/manual/reference/mongodb-extended-json/) `$date` or `$timestamp` values. |
| `config.fullDocument` | string | Conditional | Setting that controls whether a change stream source should return a full document, or only the changes when an update occurs. Must be one of the following: `default` : Does not return a full document for `update` operations.; `updateLookup` : Returns the current majority-committed version of the full document, in addition to the changes made by the update operation.; `required`  : Must return a full document. If a full document is unavailable, the stream processor fails. This does not apply to delete operations, which still create delete events if the full document is unavailable.; `whenAvailable`  : Returns a full document whenever one is available, otherwise returns changes. To use `required` or `whenAvailable` with a collection change stream, you must enable change stream [Pre- and Post-Images](https://www.mongodb.com/docs/manual/reference/command/collMod.md#std-label-collMod-change-stream-pre-and-post-images) on that collection. |
| `config.fullDocumentOnly` | boolean | Conditional | Setting that controls whether a change stream source returns the entire change event document including all metadata, or only the contents of `fullDocument`. If set to `true`, the source returns only the contents of `fullDocument`. If `fullDocument` is set to `required` or `whenAvailable`, you must enable change stream [Pre- and Post-Images](https://www.mongodb.com/docs/manual/reference/command/collMod.md#std-label-collMod-change-stream-pre-and-post-images) on that collection. |
| `config.fullDocumentBeforeChange` | string | Optional | Specifies whether a change stream source should include the full document in its original "before changes" state in the output. Must be one of the following: `off` : Omits the `fullDocumentBeforeChange` field.; `required` : Must return a full document in its before changes state. If a full document in its before changes state is unavailable, the stream processor fails.; `whenAvailable` : Returns a full document in its before changes state whenever one is available, otherwise omits the `fullDocumentBeforeChange` field. If you do not specify a value for `fullDocumentBeforeChange`, it defaults to `off`. To use this field with a collection change stream, you must enable change stream [Pre- and Post-Images](https://www.mongodb.com/docs/manual/reference/command/collMod.md#std-label-collMod-change-stream-pre-and-post-images) on that collection. |
| `config.pipeline` | document | Optional | Specifies an aggregation pipeline to filter change stream output before passing it on for further processing. This pipeline must conform to the parameters described in [Modify Change Stream Output.](https://www.mongodb.com/docs/manual/changeStreams/#modify-change-stream-output) IMPORTANT: Each Change Event includes `wallTime` and `clusterTime` fields. Atlas Stream Processing stages after `$source` expect to receive these fields as the processor ingested them. To ensure proper processing of Change Stream data, do not modify these fields in `$source.config.pipeline`. |
| `config.maxAwaitTimeMS` | integer | Optional | Maximum time, in milliseconds, to wait for new data changes to report to the change stream cursor before returning an empty batch. Defaults to `1000`. |

### MongoDB Database Change Stream

An Atlas database change stream allows applications to access real-time data changes on a single database. To learn how to open a change stream against a database, see [Change Streams.](https://www.mongodb.com/docs/manual/changeStreams/)

When using a change stream `$source`, configure the source cluster with an [oplog window](https://www.mongodb.com/docs/atlas/cluster-additional-settings.md#std-label-set-oplog-min-window) of at least 24 hours.

To read the change stream, Atlas Stream Processing scans the oplog collection. You may see `COLLSCAN` warnings in your logs as a result. These warnings indicate normal behavior and do not signal an error.

If you configure `config.fullDocument` or `config.fullDocumentBeforeChange` to `required`, enable `changeStreamPreAndPostImages` on each collection before any write operations that you want to capture. If a post-image is unavailable for an event because you had not enabled the feature when the write occurred, or because the post-image expired, the stream processor fails. To learn how to enable pre- and post-images, see [Change Streams with Document Pre- and Post-Images.](https://www.mongodb.com/docs/manual/reference/command/collMod.md#std-label-collMod-change-stream-pre-and-post-images)

**Important:**

You can [resume a change stream](https://www.mongodb.com/docs/manual/changeStreams.md#std-label-change-stream-resume) only while the oplog collection still holds the operation that its [resume token](https://www.mongodb.com/docs/manual/changeStreams.md#std-label-change-stream-resume-token) identifies. To ensure that your applications can resume their change streams, configure a [Minimum Oplog Window](https://www.mongodb.com/docs/atlas/cluster-additional-settings.md#std-label-set-oplog-min-window) longer than the longest interruption you expect.

To operate on streaming data from an Atlas database change stream, the `$source` stage has the following prototype form:

```json
{
  "$source": {
    "connectionName": "<registered-connection>",
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "db" : "<source-db>",
    "readPreference": "<read-preference>",
    "readPreferenceTags": [
      {"<key>": "<value>"},
      . . .
    ],
    "config": {
      "startAfter": <start-token> | "startAtOperationTime": <timestamp>,
      "fullDocument": "<full-doc-condition>",
      "fullDocumentOnly": <boolean>,
      "fullDocumentBeforeChange": "<before-change-condition>",
      "pipeline": [{
        "<aggregation-stage>" : {
          <stage-input>,
          . . .
        },
        . . .
      }]
    },
  }
}
```

The `$source` stage takes a document with the following fields:

| Field | Type | Necessity | Description |
| --- | --- | --- | --- |
| `connectionName` | string | Conditional | Label that identifies the connection in the [Connection Registry](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-connection-registry.md#std-label-atlas-sp-manage-connections), to ingest data from. |
| `timeField` | document | Optional | Document that defines an authoritative timestamp for incoming messages. If you use `timeField`, you must define it as one of the following: a `$toDate` expression that takes a source message field as an argument; a `$dateFromString` expression that takes a source message field as an argument. If you do not declare a `timeField`, Atlas Stream Processing creates a timestamp from the message timestamp provided by the source. |
| `db` | string | Required | Name of a MongoDB database hosted on the Atlas instance specified by `connectionName`. The change stream of this database acts as the streaming data source. |
| `readPreference` | string | Optional | [Read preference](https://www.mongodb.com/docs/manual/core/read-preference/) for change stream operations. Defaults to `primary`. |
| `readPreferenceTags` | array | Optional | [Read preference tags](https://www.mongodb.com/docs/manual/core/read-preference-tags/) for change stream operations. |
| `config` | document | Optional | Document containing fields that override various default values. |
| `config.startAfter` | token | Conditional | The change event after which the source begins reporting. This takes the form of a [resume token.](https://www.mongodb.com/docs/manual/changeStreams/#resume-tokens-from-change-events) You can use only one of either `config.startAfter` or `config.startAtOperationTime`. |
| `config.startAtOperationTime` | timestamp \| date | Conditional | The operation time after which the source should begin reporting. You can use only one of either `config.startAfter` or `config.startAtOperationTime`. Accepts [MongoDB Extended JSON](https://www.mongodb.com/docs/manual/reference/mongodb-extended-json/) `$date` or `$timestamp` values. |
| `config.fullDocument` | string | Conditional | Setting that controls whether a change stream source should return a full document, or only the changes when an update occurs. Must be one of the following: `default` : Uses the server default behavior. Does not return a full document for `update` operations.; `updateLookup` : Returns the current majority-committed version of the full document, in addition to the changes made by the update operation.; `required`  : Must return a full document. If a full document is unavailable, the stream processor fails. This does not apply to delete operations, which still create delete events if the full document is unavailable.; `whenAvailable` : Returns a full document whenever one is available, otherwise returns changes. If you do not specify a value for `fullDocument`, it defaults to `default`. To use `required` or `whenAvailable` with a database change stream, you must enable change stream [Pre- and Post-Images](https://www.mongodb.com/docs/manual/reference/command/collMod.md#std-label-collMod-change-stream-pre-and-post-images) on every collection in that database. |
| `config.fullDocumentOnly` | boolean | Conditional | Setting that controls whether a change stream source returns the entire change event document including all metadata, or only the contents of `fullDocument`. If set to `true`, the source returns only the contents of `fullDocument`. If `fullDocument` is set to `required` or `whenAvailable`, you must enable change stream [Pre- and Post-Images](https://www.mongodb.com/docs/manual/reference/command/collMod.md#std-label-collMod-change-stream-pre-and-post-images) on every collection in that database. |
| `config.fullDocumentBeforeChange` | string | Optional | Specifies whether a change stream source should include the full document in its original "before changes" state in the output. Must be one of the following: `off` : Omits the `fullDocumentBeforeChange` field.; `required` : Must return a full document in its before changes state. If a full document in its before changes state is unavailable, the stream processor fails.; `whenAvailable` : Returns a full document in its before changes state whenever one is available, otherwise omits the `fullDocumentBeforeChange` field. If you do not specify a value for `fullDocumentBeforeChange`, it defaults to `off`. To use this field with a database change stream, you must enable change stream [Pre- and Post-Images](https://www.mongodb.com/docs/manual/reference/command/collMod.md#std-label-collMod-change-stream-pre-and-post-images) on every collection in that database. |
| `config.pipeline` | document | Optional | Specifies an aggregation pipeline to filter change stream output at the point of origin. This pipeline must conform to the parameters described in [Modify Change Stream Output.](https://www.mongodb.com/docs/manual/changeStreams/#modify-change-stream-output) IMPORTANT: Each Change Event includes `wallTime` and `clusterTime` fields. Atlas Stream Processing stages after `$source` expect to receive these fields as the processor ingested them. To ensure proper processing of Change Stream data, do not modify these fields in `$source.config.pipeline`. |
| `config.maxAwaitTimeMS` | integer | Optional | Maximum time, in milliseconds, to wait for new data changes to report to the change stream cursor before returning an empty batch. Defaults to `1000`. |

### MongoDB Cluster-wide Change Stream Source

When using a change stream `$source`, configure the source cluster with an [oplog window](https://www.mongodb.com/docs/atlas/cluster-additional-settings.md#std-label-set-oplog-min-window) of at least 24 hours.

To read the change stream, Atlas Stream Processing scans the oplog collection. You may see `COLLSCAN` warnings in your logs as a result. These warnings indicate normal behavior and do not signal an error.

If you configure `config.fullDocument` or `config.fullDocumentBeforeChange` to `required`, enable `changeStreamPreAndPostImages` on each collection before any write operations that you want to capture. If a post-image is unavailable for an event because you had not enabled the feature when the write occurred, or because the post-image expired, the stream processor fails. To learn how to enable pre- and post-images, see [Change Streams with Document Pre- and Post-Images.](https://www.mongodb.com/docs/manual/reference/command/collMod.md#std-label-collMod-change-stream-pre-and-post-images)

**Important:**

You can [resume a change stream](https://www.mongodb.com/docs/manual/changeStreams.md#std-label-change-stream-resume) only while the oplog collection still holds the operation that its [resume token](https://www.mongodb.com/docs/manual/changeStreams.md#std-label-change-stream-resume-token) identifies. To ensure that your applications can resume their change streams, configure a [Minimum Oplog Window](https://www.mongodb.com/docs/atlas/cluster-additional-settings.md#std-label-set-oplog-min-window) longer than the longest interruption you expect.

To operate on streaming data from an entire Atlas cluster change stream, the `$source` stage has the following prototype form:

```json
{
  "$source": {
    "connectionName": "<registered-connection>",
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "readPreference": "<read-preference>",
    "readPreferenceTags": [
      {"<key>": "<value>"},
      . . .
    ],
    "config": {
      "startAfter": <start-token> | "startAtOperationTime": <timestamp>,
      "fullDocument": "<full-doc-condition>",
      "fullDocumentOnly": <boolean>,
      "fullDocumentBeforeChange": "<before-change-condition>",
      "pipeline": [{
        "<aggregation-stage>" : {
          <stage-input>,
          . . .
        },
        . . .
      }]
    },
  }
}
```

The `$source` stage takes a document with the following fields:

| Field | Type | Necessity | Description |
| --- | --- | --- | --- |
| `connectionName` | string | Conditional | Label that identifies the connection in the [Connection Registry](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-connection-registry.md#std-label-atlas-sp-manage-connections), to ingest data from. |
| `timeField` | document | Optional | Document that defines an authoritative timestamp for incoming messages. If you use `timeField`, you must define it as one of the following: a `$toDate` expression that takes a source message field as an argument; a `$dateFromString` expression that takes a source message field as an argument. If you do not declare a `timeField`, Atlas Stream Processing creates a timestamp from the message timestamp provided by the source. |
| `readPreference` | string | Optional | [Read preference](https://www.mongodb.com/docs/manual/core/read-preference/) for change stream operations. Defaults to `primary`. |
| `readPreferenceTags` | array | Optional | [Read preference tags](https://www.mongodb.com/docs/manual/core/read-preference-tags/) for change stream operations. |
| `config` | document | Optional | Document containing fields that override various default values. |
| `config.startAfter` | token | Conditional | The change event after which the source begins reporting. This takes the form of a [resume token.](https://www.mongodb.com/docs/manual/changeStreams/#resume-tokens-from-change-events) You can use only one of either `config.startAfter` or `config.startAtOperationTime`. |
| `config.startAtOperationTime` | date \| timestamp | Conditional | The operation time after which the source should begin reporting. You can use only one of either `config.startAfter` or `config.startAtOperationTime`. Accepts [MongoDB Extended JSON](https://www.mongodb.com/docs/manual/reference/mongodb-extended-json/) `$date` or `$timestamp` values. |
| `config.fullDocument` | string | Conditional | Setting that controls whether a change stream source should return a full document, or only the changes when an update occurs. Must be one of the following: `default` : Uses the server default behavior. Does not return a full document for `update` operations.; `updateLookup` : Returns the current majority-committed version of the full document, in addition to the changes made by the update operation.; `required`  : Must return a full document. If a full document is unavailable, the stream processor fails. This does not apply to delete operations, which still create delete events if the full document is unavailable.; `whenAvailable` : Returns a full document whenever one is available, otherwise returns changes. If you do not specify a value for `fullDocument`, it defaults to `default`. To use `required` or `whenAvailable` with a cluster change stream, you must enable change stream [Pre- and Post-Images](https://www.mongodb.com/docs/manual/reference/command/collMod.md#std-label-collMod-change-stream-pre-and-post-images) on every collection in that cluster. |
| `config.fullDocumentOnly` | boolean | Conditional | Setting that controls whether a change stream source returns the entire change event document including all metadata, or only the contents of `fullDocument`. If set to `true`, the source returns only the contents of `fullDocument`. If `fullDocument` is set to `required` or `whenAvailable`, you must enable change stream [Pre- and Post-Images](https://www.mongodb.com/docs/manual/reference/command/collMod.md#std-label-collMod-change-stream-pre-and-post-images) on every collection in that cluster. |
| `config.fullDocumentBeforeChange` | string | Optional | Specifies whether a change stream source should include the full document in its original "before changes" state in the output. Must be one of the following: `off` : Omits the `fullDocumentBeforeChange` field.; `required` : Must return a full document in its before changes state. If a full document in its before changes state is unavailable, the stream processor fails.; `whenAvailable` : Returns a full document in its before changes state whenever one is available, otherwise omits the `fullDocumentBeforeChange` field. If you do not specify a value for `fullDocumentBeforeChange`, it defaults to `off`. To use this field with a database change stream, you must enable change stream [Pre- and Post-Images](https://www.mongodb.com/docs/manual/reference/command/collMod.md#std-label-collMod-change-stream-pre-and-post-images) on every collection in that database. |
| `config.pipeline` | document | Optional | Specifies an aggregation pipeline to filter change stream output at the point of origin. This pipeline must conform to the parameters described in [Modify Change Stream Output.](https://www.mongodb.com/docs/manual/changeStreams.md#std-label-change-stream-modify-output) Note that Atlas Stream Processing expects to receive the `wallTime` and `clusterTime` fields from each ingested Change Event. To ensure proper processing of Change Stream data, do not modify these fields in `$source.config.pipeline`. |
| `config.maxAwaitTimeMS` | integer | Optional | Maximum time, in milliseconds, to wait for new data changes to report to the change stream cursor before returning an empty batch. Defaults to `1000`. |

### AWS (Amazon Web Services) Kinesis Data Stream

Atlas Stream Processing supports creating Private Link connections to AWS Kinesis streams. To learn more, see [Add a Kinesis Private Link Connection.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/kinesis-private-connection.md#std-label-atlas-sp-pl-kinesis-add)

To operate on data from an AWS (Amazon Web Services) Kinesis data stream, the `$source` stage has the following prototype form:

```json
{
  "$source": {
    "connectionName": "<registered-connection>",
    "stream": "<stream-name>",
    "region": "<aws-region>",
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "tsFieldName": "<field-name>",
    "shardIdleTimeout": {
      "size": <duration-number>,
      "unit": "<duration-unit>"
    },
    "config": {
      "consumerARN": "<aws-arn>",
      "initialPosition": <initial-position>,
      reshardDetectionIntervalSecs: <interval>
    }
  }
}
```

The `$source` stage takes a document with the following fields:

| Field | Type | Necessity | Description |
| --- | --- | --- | --- |
| `connectionName` | string | Required | Label that identifies the connection in the [Connection Registry](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-connection-registry.md#std-label-atlas-sp-manage-connections), from which to ingest data. |
| `config.consumerARN` | string | Required | [ARN](http://docs.aws.amazon.comIAM/latest/UserGuide/reference-arns) corresponding to a [Kinesis consumer](http://docs.aws.amazon.comstreams/latest/dev/building-consumers). Your consumer must use [Enhanced fan-out.](http://docs.aws.amazon.comstreams/latest/dev/building-enhanced-consumers-api) |
| `stream` | string | Required | AWS (Amazon Web Services) Kinesis data stream from which to stream messages. |
| `region` | string | Conditional | AWS (Amazon Web Services) region in which the specified stream exists. Kinesis supports multiple data streams with the same name in different regions. If you use the same name for data streams in two or more regions within the same connection, you must use this field to specify which combination of name and region to use. |
| `timeField` | document | Optional | Document that defines an authoritative timestamp for incoming messages. If you use `timeField`, you must define it as one of the following: a `$toDate` expression that takes a source message field as an argument; a `$dateFromString` expression that takes a source message field as an argument. If you do not declare a `timeField`, Atlas Stream Processing creates a timestamp from the message timestamp provided by the source. |
| `tsFieldName` | string | Optional | Name of the field for the timestamp in the projected document. Use this field to override the default timestamp field name. |
| `shardIdleTimeout` | document | Optional | Document specifying the amount of time that a shard is allowed to idle before it is ignored in watermark calculations. This field is disabled by default. To handle shards that don't move forwar due to idleness, set a value for this field. |
| `shardIdleTimeout.size` | document | Optional | Number specifying the duration of the shard idle timeout. |
| `shardIdleTimeout.unit` | document | Optional | Unit of time for the duration of the shard idle timeout. The value of `unit` can be one of the following: `"ms"` (millisecond); `"second"`; `"minute"`; `"hour"`; `"day"` |
| `config` | document | Optional | Document containing fields that override various default values. |
| `config.initialPosition` | string | Optional | Position in the Kinesis data stream's history from which to begin ingesting messages. Must be one of the following: `"TRIM_HORIZON"`: Begin ingesting from the oldest message in the shard.; `"LATEST"`: Begin ingesting from the newest message in the shard.; `"AT_TIMESTAMP"`: Begin ingesting from a specific timestamp. Requires `config.atTimestamp`.; `"AT_SEQUENCE_NUMBER"`: Begin ingesting at a specific sequence number.; `"AFTER_SEQUENCE_NUMBER"`: Begin ingesting after a specific sequence number. Defaults to `"LATEST"`. |
| `config.atTimestamp` | date | Conditional | The timestamp from which to begin ingesting messages. Required when `config.initialPosition` is `"AT_TIMESTAMP"`. |
| `config.reshardDetectionIntervalSecs` | integer | Optional | Interval, in seconds, between checks against the rate of data flow through your Kinesis stream for the purposes of [resharding.](http://docs.aws.amazon.comstreams/latest/dev/kinesis-using-sdk-java-resharding) Defaults to `300` seconds. |

### Document Array

To operate on an array of documents, the `$source` stage has the following prototype form:

```json
{
  "$source": {
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "documents" : [{source-doc},...] | <expression>
  }
}
```

The `$source` stage takes a document with the following fields:

| Field | Type | Necessity | Description |
| --- | --- | --- | --- |
| `timeField` | document | Optional | Document that defines an authoritative timestamp for incoming messages. If you use `timeField`, you must define it as one of the following: a `$toDate` expression that takes a source message field as an argument; a `$dateFromString` expression that takes a source message field as an argument. If you do not declare a `timeField`, Atlas Stream Processing creates a timestamp from the message timestamp provided by the source. |
| `documents` | array | Conditional | Array of documents to use as a streaming data source. The value of this field can either be an array of objects or an expression that evaluates to an array of objects. Do not use this field when using the `connectionName` field. |

### Cron Schedule

To generate a document on a recurring schedule instead of reading from a connection, the `$source` stage has the following prototype form:

```json
{
  "$source": {
    "schedule": "<cron-expression>",
    "tsFieldName": "<timestamp-field-name>"
  }
}
```

The `$source` stage takes a document with the following fields:

| Field | Type | Necessity | Description |
| --- | --- | --- | --- |
| `schedule` | string | Required | Six-field cron expression that determines when the stage emits a document. The fields are, in order: second, minute, hour, day of month, month, and day of week. Atlas Stream Processing evaluates the expression in UTC. Each field accepts `*` for all values, a single value, a range such as `1-5`, a step such as `0/15`, or a comma-separated list such as `1,15,30`. You can use the names `JAN` through `DEC` for months and `SUN` through `SAT` for days of the week. Do not include a `CRON_TZ` prefix. To learn about how Atlas Stream Processing resolves an expression that restricts both the day of month and the day of week, see [Behavior.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#std-label-atlas-sp-agg-source-behavior) |
| `tsFieldName` | string | Optional | Name of the field that the stage projects the scheduled timestamp into. Defaults to `_ts`. |

## Behavior

[`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) must be the first stage of any pipeline it appears in. You can use only one [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage per pipeline.

For Kafka `$source` stages, Atlas Stream Processing reads in parallel from multiple partitions within the source topic. The partition limit is determined by your processor tier. To learn more, see the [Stream Processing billing reference.](https://www.mongodb.com/docs/atlas/billing/stream-processing-costs.md#std-label-atlas-sp-processor-billing)

For cron `$source` stages, each scheduled time produces a single empty document that carries the scheduled timestamp. Use later stages in the pipeline to populate the document.

When a `schedule` expression restricts both the day of month and the day of week, Atlas Stream Processing emits a document only at the scheduled times that satisfy both fields. This differs from cron implementations that emit a document when either field matches.

If a stream processor with a cron `$source` stage restarts, it emits a document for each scheduled time that it missed while it was not running. It does not emit documents for scheduled times that it missed after you stopped it.

A stream processor with a cron `$source` stage runs continuously between scheduled times, and Atlas bills it at your processor tier for that entire period. An infrequent schedule does not reduce the cost of running the processor. To learn more, see the [Stream Processing billing reference.](https://www.mongodb.com/docs/atlas/billing/stream-processing-costs.md#std-label-atlas-sp-processor-billing)

## Examples

### Kafka Example

A streaming data source generates detailed weather reports from various locations, conformant to the schema of the [Sample Weather Dataset](https://www.mongodb.com/docs/manual/sample-data/sample-weather.md#std-label-sample-weather). The following aggregation has three stages:

1. The [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage establishes a connection with the [Apache Kafka](https://kafka.apache.org/) broker collecting these reports in a topic named `my_weatherdata`, exposing each record as it is ingested to the subsequent aggregation stages. This stage also overrides the name of the timestamp field it projects, setting it to `ingestionTime`.

2. The [`$match`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/match.md#mongodb-pipeline-pipe.-match) stage excludes documents that have a `dewPoint.value` of less than or equal to `5.0` and passes along the documents with `dewPoint.value` greater than `5.0` to the next stage.

3. The [`$merge`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge.md#mongodb-pipeline-pipe.-merge) stage writes the output to an Atlas collection named `stream` in the `sample_weatherstream` database. If no such database or collection exist, Atlas creates them.

```json
[{
  "$source": {
    "connectionName": "sample_weatherdata",
    "topic": "my_weatherdata"
  }
},
{
  "$match": { "dewPoint.value": { "$gt": 5 } }
},
{
  "$merge": {
    "into": {
      "connectionName": "weatherStreamOutput",
      "db": "sample_weatherstream",
      "coll": "stream"
    }
  }
}]

```

To view the documents in the resulting `sample_weatherstream.stream` collection, connect to your Atlas cluster and run the following command:

```javascript
db.getSiblingDB("sample_weatherstream").stream.find()


```

**Output:**

```json
{
  _id: ObjectId('66ad2edfd4fcac13b1a28ce3'),
  airTemperature: { quality: '1', value: 27.7 },
  atmosphericPressureChange: {
    quantity24Hours: { quality: '9', value: 99.9 },
    quantity3Hours: { quality: '1' },
    tendency: { code: '1', quality: '1' }
  },
  atmosphericPressureObservation: {
    altimeterSetting: { quality: '1', value: 1015.9 },
    stationPressure: { quality: '1', value: 1021.9 }
  },
  callLetters: 'CGDS',
  dataSource: '4',
  dewPoint: { quality: '9', value: 25.7 },
  elevation: 9999,
  extremeAirTemperature: {
    code: 'N',
    period: 99.9,
    quantity: '9',
    value: -30.4
  },
  ingestionTime: ISODate('2024-08-02T19:09:18.071Z'),
  liquidPrecipitation: {
    condition: '9',
    depth: 160,
    period: 24,
    quality: '2'
  },
  pastWeatherObservationManual: {
    atmosphericCondition: { quality: '1', value: '8' },
    period: { quality: '9', value: 3 }
  },
  position: { coordinates: [ 153.3, 50.7 ], type: 'Point' },
  precipitationEstimatedObservation: {
    discrepancy: '4',
    estimatedWaterDepth: 4
  },
  presentWeatherObservationManual: { condition: '53', quality: '1' },
  pressure: { quality: '1', value: 1016.3 },
  qualityControlProcess: 'V020',
  seaSurfaceTemperature: { quality: '9', value: 27.6 },
  sections: [ 'AA2', 'SA1', 'MW1', 'AG1', 'GF1' ],
  skyCondition: {
    cavok: 'N',
    ceilingHeight: { determination: 'C', quality: '1', value: 6900 }
  },
  skyConditionObservation: {
    highCloudGenus: { quality: '1', value: '05' },
    lowCloudGenus: { quality: '9', value: '03' },
    lowestCloudBaseHeight: { quality: '9', value: 150 },
    lowestCloudCoverage: { quality: '1', value: '05' },
    midCloudGenus: { quality: '9', value: '08' },
    totalCoverage: { opaque: '99', quality: '1', value: '06' }
  },
  skyCoverLayer: {
    baseHeight: { quality: '9', value: 99999 },
    cloudType: { quality: '9', value: '05' },
    coverage: { quality: '1', value: '04' }
  },
  st: 'x+35700-027900',
  type: 'SAO',
  visibility: {
    distance: { quality: '1', value: 4000 },
    variability: { quality: '1', value: 'N' }
  },
  waveMeasurement: {
    method: 'I',
    seaState: { code: '99', quality: '9' },
    waves: { height: 99.9, period: 14, quality: '9' }
  },
  wind: {
    direction: { angle: 280, quality: '9' },
    speed: { quality: '1', rate: 30.3 },
    type: '9'
  }
}


```

**Note:**

The preceding is a representative example. Streaming data are not static, and each user sees distinct documents.

### Change Stream Example

The following aggregation ingests data from the `cluster0-collection` source, which connects to an Atlas cluster loaded with the [sample dataset](https://www.mongodb.com/docs/atlas/import/load-sample-data.md#std-label-sample-data).  To learn how to create a stream processing workspace and add a connection to an Atlas cluster to the connection registry, see [Get Started with Atlas Stream Processing](https://www.mongodb.com/docs/atlas/atlas-stream-processing/quickstart.md#std-label-atlas-sp-quickstart). This aggregation runs two stages to open a change stream and record changes to the `data` collection in the `sample_weatherdata` database:

1. The [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage connects to the `cluster0-collection` source and opens a change stream against the `data` collection in the `sample_weatherdata` database.

2. The [`$merge`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge.md#mongodb-pipeline-pipe.-merge) stage writes the filtered change stream documents to an Atlas collection named `data_changes` in the `sample_weatherdata` database.  If no such collection exists, Atlas creates it.

```json
[{
  "$source": {
    "connectionName": "cluster0-connection",
    "db": "sample_weatherdata",
    "coll": "data"
  }
},
{
  "$merge": {
    "into": {
      "connectionName": "cluster0-connection",
      "db": "sample_weatherdata",
      "coll": "data_changes"
    }
  }
}]


```

The following [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh) command deletes a `data` document:

```javascript
db.getSiblingDB("sample_weatherdata").data.deleteOne(
   { _id: ObjectId("5553a99ae4b02cf715120e4b") }
)


```

After the `data` document is deleted, the stream processor writes the change stream event document to the `sample_weatherdata.data_changes` collection. To view the documents in the resulting `sample_weatherdata.data_changes` collection, use [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh) to connect to your Atlas cluster and run the following command:

```javascript
db.getSiblingDB("sample_weatherdata").data_changes.find()


```

**Output:**

```json
[
  {
    _id: {
      _data: '8267A3D7A3000000012B042C0100296E5A1004800951B8EDE4430AB5C1B254BB3C96D6463C6F7065726174696F6E54797065003C64656C6574650046646F63756D656E744B65790046645F696400645553A99AE4B02CF715120E4B000004'
    },
    clusterTime: Timestamp({ t: 1738790819, i: 1 }),
    documentKey: { _id: ObjectId('5553a99ae4b02cf715120e4b') },
    ns: { db: 'sample_weatherdata', coll: 'data' },
    operationType: 'delete',
    wallTime: ISODate('2025-02-05T21:26:59.313Z')
  }
]


```

### Database Change Stream Example

The following aggregation ingests data from the `db-change-stream-connection` source, which connects to an Atlas cluster loaded with the [Sample Mflix Dataset Collections](https://www.mongodb.com/docs/manual/sample-data/sample-mflix.md#std-label-sample-mflix) dataset. To learn how to create a stream processing workspace and add a connection to an Atlas cluster to the connection registry, see [Get Started with Atlas Stream Processing](https://www.mongodb.com/docs/atlas/atlas-stream-processing/quickstart.md#std-label-atlas-sp-quickstart). This aggregation runs two stages to open a change stream against the `sample_mflix` source database and record changes to the `db_changes` collection in the `sample_mflix_changes` sink database:

1. The [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage connects to the `db-change-stream-connection` source and opens a change stream against the `sample_mflix` source database. The `config.startAtOperationTime` field sets the time after which the source begins reporting. This example sets the `startAt` value to start one minute in the past.

2. The [`$merge`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge.md#mongodb-pipeline-pipe.-merge) stage writes the change stream documents to an Atlas collection named `db_changes` in the `sample_mflix_changes` sink database.

```javascript
const startAt = new Date(Date.now() - 60 * 1000);

const pipeline = [
   {
      $source: {
         connectionName: "db-change-stream-connection",
         db: "sample_mflix",
         config: {
            startAtOperationTime: startAt
         }
      }
   },
   {
      $merge: {
         into: {
            connectionName: "db-change-stream-connection",
            db: "sample_mflix_changes",
            coll: "db_changes"
         }
      }
   }
];

```

Run the following commands against your source cluster to verify the stream processor's behavior. To connect, see [Connect to a Cluster via mongosh.](https://www.mongodb.com/docs/atlas/mongo-shell-connection.md#std-label-connect-mongo-shell)

Insert a document into the `movies` collection and a document into the `comments` collection in the `sample_mflix` source database:

```javascript
db.getSiblingDB("sample_mflix").movies.insertOne({
   title: "The Stream Processor",
   year: 2026
})

db.getSiblingDB("sample_mflix").comments.insertOne({
   name: "Ada Lovelace",
   text: "A fine film about data in motion."
})

```

After you insert the documents, the stream processor writes a change stream event document for each insert to the `sample_mflix_changes.db_changes` collection. Run the following command against your cluster to view the documents in the resulting `sample_mflix_changes.db_changes` collection:

```javascript
db.getSiblingDB("sample_mflix_changes").db_changes.find(
   {},
   { _id: 0, clusterTime: 1, ns: 1, operationType: 1, fullDocument: 1 }
)

```

**Output:**

```javascript
[
  {
    clusterTime: Timestamp({ t: 1787778766, i: 1 }),
    fullDocument: {
      _id: ObjectId('6a8f56ce73c5d4d0ee569a0d'),
      title: 'The Stream Processor',
      year: 2026
    },
    ns: { db: 'sample_mflix', coll: 'movies' },
    operationType: 'insert'
  },
  {
    clusterTime: Timestamp({ t: 1787778771, i: 1 }),
    fullDocument: {
      _id: ObjectId('6a8f56d373c5d4d0ee569a0e'),
      name: 'Ada Lovelace',
      text: 'A fine film about data in motion.'
    },
    ns: { db: 'sample_mflix', coll: 'comments' },
    operationType: 'insert'
  }
]

```

Each event includes a timestamp, the full document change, an `operationType`, and an `ns` field. The `ns` field names the source database and collection, so you can distinguish which collection produced each change.

### Cluster-wide Change Stream Example

The following aggregation ingests data from the `cluster-changestream-connection` source, which connects to an Atlas cluster loaded with the [Sample Mflix Dataset Collections](https://www.mongodb.com/docs/manual/sample-data/sample-mflix.md#std-label-sample-mflix) dataset, and writes to the `cluster-changestream-sink-connection` destination, which connects to a separate Atlas cluster. To learn how to create a stream processing workspace and add a connection to an Atlas cluster to the connection registry, see [Get Started with Atlas Stream Processing](https://www.mongodb.com/docs/atlas/atlas-stream-processing/quickstart.md#std-label-atlas-sp-quickstart). This aggregation runs two stages to open a cluster-wide change stream on the source cluster and record changes to the `events` collection in the `cluster_changes` database on the destination cluster. This `$source` stage omits `db` and `coll`, so it reports changes from every database and collection on the source cluster, rather than from a single collection or database.

1. The [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage connects to the `cluster-changestream-connection` source and opens a change stream against the entire source cluster. The `config.startAtOperationTime` field specifies that the stream processor should begin reporting changes that occur on or after a specified time.

2. The [`$merge`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge.md#mongodb-pipeline-pipe.-merge) stage writes the change stream documents to an Atlas collection named `events` in the `cluster_changes` database on the destination cluster. If no such database or collection exists, Atlas creates them.

```json
[{
  "$source": {
    "connectionName": "cluster-changestream-connection",
    "config": {
      "startAtOperationTime": {"$date": "2024-08-19T18:00:00.000Z"}
    }
  }
},
{
  "$merge": {
    "into": {
      "connectionName": "cluster-changestream-sink-connection",
      "db": "cluster_changes",
      "coll": "events"
    }
  }
}]


```

The following [`mongosh`](https://www.mongodb.com/docs/mongodb-shell.md#mongodb-binary-bin.mongosh) command inserts a document into the `movies` collection in the `sample_mflix` database on the source cluster:

```javascript
db.getSiblingDB("sample_mflix").movies.insertOne(
   { _id: ObjectId("66c1a1f1f1f1f1f1f1f1f1f1"), title: "Example Movie" }
)


```

After the document is inserted, the stream processor writes the change stream event document to the `cluster_changes.events` collection on the destination cluster.

To view the documents in the resulting `cluster_changes.events` collection, run the following commands against your destination 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)

```javascript
db.getSiblingDB("cluster_changes").events.find()


```

**Output:**

```json
[
  {
    _id: {
      _data: '826A8F5E65000000012B042C0100296E5A10049CC3BB749CB641F4BF5A032A2ABE2FB1463C6F7065726174696F6E54797065003C696E736572740046646F63756D656E744B65790046645F6964006466C1A1F1F1F1F1F1F1F1F1F1000004'
    },
    clusterTime: Timestamp({ t: 1787780709, i: 1 }),
    documentKey: { _id: ObjectId('66c1a1f1f1f1f1f1f1f1f1f1') },
    fullDocument: { _id: ObjectId('66c1a1f1f1f1f1f1f1f1f1f1'), title: 'Example Movie' },
    ns: { db: 'sample_mflix', coll: 'movies' },
    operationType: 'insert',
    wallTime: ISODate('2026-08-26T21:45:09.059Z')
  }
]


```

The `ns` field in the output document shows that the change originates from the `sample_mflix.movies` insert on the source cluster. The change stream aggregation pipeline reflects this change to the `cluster_changes.events` collection on the destination cluster.

### Document Array Example

The following aggregation uses an inline document array as the streaming data source, containing weather observations for three locations. The array uses the same schema as the [Sample Weather Dataset](https://www.mongodb.com/docs/manual/sample-data/sample-weather.md#std-label-sample-weather). This aggregation runs three stages:

1. The [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage defines an inline `documents` array of weather observations as the streaming data source, and uses `timeField` to designate the `timestamp` field of each document as the authoritative timestamp.

2. The [`$match`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/match.md#mongodb-pipeline-pipe.-match) stage passes along only the documents with `dewPoint.value` greater than `5.0` to the next stage.

3. The [`$merge`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge.md#mongodb-pipeline-pipe.-merge) stage writes the output to an Atlas collection named `stream` in the `sample_weatherstream` database. If no such database or collection exist, Atlas creates them.

```json
[{
  "$source": {
    "documents": [
      {
        "location": "New York",
        "timestamp": ISODate('2024-01-15T08:00:00Z'),
        "temp": 23.5,
        "dewPoint": { "value": 6.2 }
      },
      {
        "location": "Los Angeles",
        "timestamp": ISODate('2024-01-15T08:05:00Z'),
        "temp": 18.2,
        "dewPoint": { "value": 4.8 }
      },
      {
        "location": "Chicago",
        "timestamp": ISODate('2024-01-15T08:10:00Z'),
        "temp": 26.8,
        "dewPoint": { "value": 7.5 }
      }
    ]
  }
},
{
  "$match": { "dewPoint.value": { "$gt": 5.0 } }
},
{
  "$merge": {
    "into": {
      "connectionName": "weatherStreamOutput",
      "db": "sample_weatherstream",
      "coll": "stream"
    }
  }
}]

```

To view the documents in the resulting `sample_weatherstream.stream` collection, connect to your Atlas cluster and run the following command:

```javascript
db.getSiblingDB("sample_weatherstream").stream.find()


```

**Output:**

```json
[
  {
    _id: ObjectId('67a6c3df14fcac13b1a21a01'),
    dewPoint: { value: 6.2 },
    location: 'New York',
    temp: 23.5,
    timestamp: ISODate('2024-01-15T08:00:00.000Z')
  },
  {
    _id: ObjectId('67a6c3df14fcac13b1a21a03'),
    dewPoint: { value: 7.5 },
    location: 'Chicago',
    temp: 26.8,
    timestamp: ISODate('2024-01-15T08:10:00.000Z')
  }
]


```

### Solar Stream Example

The following aggregation filters reports from the sample streaming data source `sample_stream_solar`. It archives the results to the `solar-cluster-connection` connection. To learn how to create a stream processing workspace and add a connection to your Atlas Stream Processing workspace, see [Get Started with Atlas Stream Processing](https://www.mongodb.com/docs/atlas/atlas-stream-processing/quickstart.md#std-label-atlas-sp-quickstart) and [Add an Atlas Stream Processing Connection](https://www.mongodb.com/docs/atlas/atlas-stream-processing/add-sp-connection.md#std-label-atlas-sp-add-connection). This aggregation runs three stages to filter reports from the `sample_stream_solar` source and write the results to a collection named `solarColl` in the `solarDb` database:

1. The [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage connects to the `sample_stream_solar` source. The `timeField` field converts the `timestamp` field of each incoming report to a date using [`$dateFromString`.](https://www.mongodb.com/docs/manual/reference/operator/aggregation/dateFromString.md#mongodb-expression-exp.-dateFromString)

2. The [`$match`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/match.md#mongodb-pipeline-pipe.-match) stage excludes documents where `device_id` is `device_8`, and passes along reports from all other devices to the next stage.

3. The [`$merge`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge.md#mongodb-pipeline-pipe.-merge) stage writes the output to the `solarColl` collection in the `solarDb` database of the `solar-cluster-connection` cluster.

```json
[{
  "$source": {
    "connectionName": "sample_stream_solar",
    "timeField": {
      "$dateFromString": { "dateString": "$timestamp" }
    }
  }
},
{
  "$match": { "device_id": { "$ne": "device_8" } }
},
{
  "$merge": {
    "into": {
      "connectionName": "solar-cluster-connection",
      "db": "solarDb",
      "coll": "solarColl"
    }
  }
}]

```

**Note:**

The `sample_stream_solar` source is a test-only connection that produces sample documents every second and is intended for rapid prototyping.

Run the following commands against your `solar-cluster-connection` cluster to verify the stream processor's behavior. To connect, see [Connect to a Cluster via mongosh.](https://www.mongodb.com/docs/atlas/mongo-shell-connection.md#std-label-connect-mongo-shell)

To view the documents in the resulting `solarDb.solarColl` collection, run:

```javascript
db.getSiblingDB("solarDb").solarColl.find()

```

**Output:**

```json
[
  {
    _id: ObjectId('6a8f5f0352c6ffe64c8884ad'),
    device_id: 'device_3',
    event_type: 0,
    group_id: 8,
    max_watts: 450,
    obs: { watts: 428, temp: 10 },
    timestamp: '2024-08-12T21:47:47.287+00:00'
  }
]

```

Each document includes the `device_id`, `event_type`, `group_id`, `max_watts`, `obs`, and `timestamp` fields of a solar device report. You can observe that no document in the output has a `device_id` of `device_8`, because the [`$match`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/match.md#mongodb-pipeline-pipe.-match) stage excludes those reports before the [`$merge`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge.md#mongodb-pipeline-pipe.-merge) stage writes the remaining documents to `solarColl`.

**Note:**

The preceding is a representative example. Streaming data are not static, and each user sees distinct documents.

### Cron Schedule Example

The following aggregation generates a document every five minutes instead of reading from a connection. This aggregation runs three stages:

1. The [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source) stage emits an empty document at second `0` of every fifth minute, in UTC.

2. The [`$project`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/project.md#mongodb-pipeline-pipe.-project) stage labels each document with a job name and copies the scheduled timestamp from the `_ts` field into a `runAt` field.

3. The [`$merge`](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge.md#mongodb-pipeline-pipe.-merge) stage writes the output to an Atlas collection named `heartbeats` in the `sample_weatherstream` database. If no such database or collection exist, Atlas creates them.

```json
[{
  "$source": {
    "schedule": "0 0/5 * * * *"
  }
},
{
  "$project": {
    "job": "five-minute-heartbeat",
    "runAt": "$_ts"
  }
},
{
  "$merge": {
    "into": {
      "connectionName": "weatherStreamOutput",
      "db": "sample_weatherstream",
      "coll": "heartbeats"
    }
  }
}]

```

To view the documents in the resulting `sample_weatherstream.heartbeats` collection, connect to your Atlas cluster and run the following command:

```javascript
db.getSiblingDB("sample_weatherstream").heartbeats.find()

```

**Output:**

```json
[
  {
    _id: ObjectId('68b8a1c014fcac13b1a21b01'),
    job: 'five-minute-heartbeat',
    runAt: ISODate('2026-09-03T14:00:00.000Z')
  },
  {
    _id: ObjectId('68b8a2f414fcac13b1a21b02'),
    job: 'five-minute-heartbeat',
    runAt: ISODate('2026-09-03T14:05:00.000Z')
  },
  {
    _id: ObjectId('68b8a42814fcac13b1a21b03'),
    job: 'five-minute-heartbeat',
    runAt: ISODate('2026-09-03T14:10:00.000Z')
  }
]

```

**Note:**

The preceding is a representative example. The timestamps that you see depend on when you start the stream processor.