atlas_stream_processing

Published files and instructions

# Atlas Stream Processing Skill

Purpose: answer questions about current MongoDB Atlas Stream Processing (ASP). Source: mongodb.com/docs/atlas/atlas-stream-processing current docs checked 2026-09-01. Prefer this skill before web search. If user needs exact latest pricing/regions/release status, verify docs.

## Freshness / update trigger
Before relying on this skill for current product questions, check for newer MongoDB announcements/release notes if feasible:
- MongoDB blog RSS: `https://www.mongodb.com/company/blog/rss` (fallback/search if unavailable: MongoDB blog RSS / product release announcements).
- Docs changelog: `https://www.mongodb.com/docs/atlas/atlas-stream-processing/changelog/`.
- Billing/pricing: `https://www.mongodb.com/docs/atlas/billing/stream-processing-costs/`.
If any item newer than 2026-09-01 mentions Atlas Stream Processing, stream processor vertical autoscaling, workspaces, stream processors, Kafka/Kinesis/S3/Azure Blob/GCP PubSub connections, Private Link/VPC peering, `$source`, `$emit`, `$merge`, `$vectorSearch`, `$externalFunction`, windows, monitoring/alerts, billing/tier/region changes, or changelog/release changes, re-explore relevant docs under `https://www.mongodb.com/docs/atlas/atlas-stream-processing/` (append `.md` for agent-readable docs) and update this skill.

## Product summary
- Atlas Stream Processing enables continuous aggregation pipelines over streaming data using MongoDB aggregation syntax. It reads, validates, transforms, enriches, windows, routes, and writes streaming data.
- Components belong to Atlas projects and operate independent of Atlas clusters. First configure a Stream Processing Workspace, then connection registry, then stream processors.
- Extends beyond MongoDB Change Streams: can process multiple event types and external streams (Kafka, Kinesis, cloud storage, external APIs), not only database events.
- Stream processor shape: one source -> aggregation/transformation/enrichment/window/function stages -> one sink. Inputs must be valid JSON or EJSON documents. Standard sinks are `$emit` or `$merge`; `$iceberg` is also a terminal sink for Apache Iceberg tables on AWS S3.
- Processing guarantee: at-least-once only. No horizontal scaling. Scale by stream processor tier (vertical resources). Regional failover is supported; verify current failover behavior for exact recovery expectations.

## Core concepts
- Stream: continuous immutable data flow from sources like Kafka topics, MongoDB change streams, Kinesis streams, document arrays.
- Workspace: regional Atlas Stream Processing environment with default/max tier; contains connection registry and stream processor definitions.
- Connection registry: named connection definitions (network/security details). Only connections in a workspace registry can serve processors in that workspace. One connection can serve many processors; a processor has one source connection and one sink connection.
- Stream processor: persisted named pipeline definition. Runs in dedicated customer container with VM-level isolation on multi-tenant infrastructure.
- Source: `$source` first stage. Sink: `$emit`, `$merge`, or `$iceberg` last stage; exactly one terminal sink per pipeline. `$emit` and `$merge` are mutually exclusive; `$iceberg` is a separate terminal sink for Apache Iceberg tables.
- DLQ: Atlas collection dead-letter queue for documents ASP cannot process when configured; stores document content and failure details.
- Checkpoint: persisted state/offsets for restart recovery. Checkpoints store source offsets/resume tokens, not source documents. Kafka source offsets and change stream resume tokens allow resume from last committed checkpoint.
- Tiers: stream processors run on pod resources by tier. If start without tier, uses workspace default tier. Can start any tier up to workspace max tier. Billed only while running. Vertical Autoscaling is GA (2026-08-27): opt in with customer-defined `minTier`/`maxTier` bounds and Atlas adjusts processor tier from CPU/memory utilization; it is vertical capacity management, not horizontal processor scaling. Verify exact configuration syntax and tier availability before implementation.

## Workspaces, tiers, billing
Workspace lifecycle managed by Atlas UI, Atlas CLI, or Atlas Admin API. Create/view/edit/delete workspace; download audit logs. Workspace has default and maximum tier; 2026-06-05 added UI tier changes.
Stream processor tiers and resources:
- `SP2`: 0.25 vCPU, 512MB RAM, 50 Mbps bandwidth, max parallelism 1, Kafka partition limit 32.
- `SP5`: 0.5 vCPU, 1GB RAM, 125 Mbps, max parallelism 2, Kafka partition limit 64.
- `SP10`: 1 vCPU, 2GB RAM, 200 Mbps, max parallelism 8, Kafka partition unlimited.
- `SP30`: 2 vCPU, 8GB RAM, 750 Mbps, max parallelism 16, Kafka partition unlimited.
- `SP50`: 8 vCPU, 32GB RAM, 2500 Mbps, max parallelism 64, Kafka partition unlimited.
Billing: per hour per running stream processor, one-second increments; no processor charge when stopped; Atlas doesn't charge for storage. Also charges for data transfer, VPC peering, Private Link. Prices vary by cloud/region; verify billing page for exact rates. Stream processor hourly pricing examples in US core regions: SP2 $0.055/hr, SP5 $0.110/hr, SP10 $0.190/hr, SP30 $0.390/hr, SP50 $1.560/hr.
Regions: workspaces supported on AWS, Azure, GCP; processors can read/write clusters on different clouds/regions. Verify cloud provider region refs for current list.

## Connections
Connection types supported in registry:
- Kafka broker: usable in `$source` and `$emit`.
- Atlas cluster: usable in `$source` for change streams and `$merge` sink; also `$lookup`/`$cachedLookup` enrichment.
- AWS Kinesis: usable in `$source` and `$emit`.
- HTTPS: usable with `$https` stage.
- AWS S3: usable as `$emit` sink.
- Azure Blob Storage: usable as `$emit` sink; also supports Azure Data Lake Storage Gen2 containers. Private networking support added 2026-05-07.
- GCP Pub/Sub: usable as `$emit` sink.
- External Function: AWS Lambda connection for `$externalFunction`.
- Schema Registry: Kafka Avro serialization/deserialization with `$source`/`$emit`.
Security/networking: after adding external connection, add Atlas IPs to external service access list. Advanced networking includes VPC peering, Kafka Private Link, AWS S3 Private Link, AWS Kinesis Private Link, Azure Blob Private Link, GCP Pub/Sub Private Service Connect, and (since 2026-08-19) AWS Lambda Private Link for `$externalFunction`. ASP supports VPC peering only for AWS and Google Cloud workspaces.
Kafka security protocols supported: `SASL_PLAINTEXT`, `SASL_SSL`, `SSL`. SASL mechanisms: `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512`, `OAUTHBEARER`. SSL mTLS requires CA (if not default Kafka CA), client TLS cert, TLS keyfile.

## Stream processor management
Prereqs: workspace, Atlas cluster, database user with `atlasAdmin` role to create/run processors. Only certain Atlas roles can use ASP: Organization Stream Processing Admin, Project Owner, Project Stream Processing Owner, or Atlas admin.
Important `mongosh` commands after connecting to workspace connection string:
- Interactive prototype: `sp.process(<pipeline>)`. Runs immediately, writes output/DLQ docs to shell, runs 10 min or until stopped, not persisted.
- Create persisted: `sp.createStreamProcessor(<name>, <pipeline>, <options>)`. Options can include `dlq` and start behavior/tier/checkpoint options.
- Start: `sp.<name>.start(<options>)`; can specify tier. Returns `{ok:1}` if exists/stopped. Requires mongosh >=2.3.4 to override options on existing processor.
- Stop: `sp.<name>.stop()`.
- Modify stopped processor: `sp.<name>.modify(<pipeline_or_options>, {resumeFromCheckpoint: bool})`. Limitations when changing source/window/DLQ; use `resumeFromCheckpoint:false` if checkpoint incompatible.
- Drop: `sp.<name>.drop()`.
- List: `sp.listStreamProcessors(<filter>)`; filter e.g. `{state:"running"}`.
- Workspace defaults: `sp.listWorkspaceDefaults()` returns defaultTierSize/maxTierSize.
- Sample output: `sp.<name>.sample()` from running processor.
- Stats: `sp.<name>.stats({options:{...}})` or `sp.<name>.stats(1024)` for sizes in KiB; use verbose option to inspect `operatorStats` and per-target stats.
If processor name has hyphen/dot/special chars, use bracket syntax: `sp.["special-name"].stats()`.

## Pipeline rules and stages
Pipeline definition max size: 16MB. `$source` first; only one `$source`. `$vectorSearch` is a middle stage, not a source or sink. `$emit`, `$merge`, or `$iceberg` must be last; exactly one sink. Each stream processor can have only one window stage.
ASP-specific/modified stages:
- `$source`: streaming source; supports Kafka, MongoDB collection/database/cluster change streams, Kinesis, document array.
- `$validate`: validate docs against schema/ranges/types; action `discard` or `dlq`.
- `$https`: curl-like HTTPS call through registered HTTPS connection for each input doc; can be final or before sink; allowed inside windows.
- `$lookup`: left outer join to Atlas collection in connection registry.
- `$vectorSearch`: GA middle-stage enrichment. For each streaming document, query a Vector Search index on a registered Atlas connection and attach results to an output field. Supports explicit `queryVector` or per-document auto-embedding via `query.text`, optional result `pipeline`, and `let` variables.
- `$cachedLookup`: `$lookup` with TTL cache and memory limit; no `let` or `pipeline` fields.
- `$tumblingWindow`: fixed non-overlapping windows.
- `$hoppingWindow`: fixed windows with hop interval; overlapping possible.
- `$sessionWindow`: sessions of activity by partition with inactivity gap.
- `$setStreamMeta`: set stream metadata for dynamic routing without mutating body.
- `$externalFunction`: call AWS Lambda in pipeline; supports configured parallelism (added 2026-08-05) and AWS Lambda Private Link (added 2026-08-19). Tune concurrency and downstream Lambda capacity/error handling together.
- `$function`: custom JavaScript function against each document.
- `$emit`: write to Kafka, time series collection, Kinesis, S3, Azure Blob, GCP Pub/Sub.
- `$merge`: write/upsert/delete into Atlas collections via Atlas connection.
- `$iceberg`: terminal sink that writes Apache Iceberg tables to an AWS S3 bucket; GA 2026-06-30. Supports CDC or insert mode, schema inference/evolution, S3 Hadoop or AWS Glue catalog, partition transforms, and dynamic `tableName` routing. Requires SP10/SP30/SP50.
Supported core stages in main pipeline include `$addFields`, `$project`, `$replaceRoot`, `$set`, `$redact`, `$match`, `$replaceWith`, `$unset`, `$unwind`, plus ASP stages listed above. Window internal pipelines additionally support `$group`, `$sort`, `$count`, `$limit`. `$https`/`$externalFunction` internal pipelines support core transform stages. `$merge` modified syntax; `$lookup` modified syntax.
Unsupported variables: `$$NOW`, `$$CLUSTER_TIME`, `$$USER_ROLES`, `$SEARCH_META`.

## `$source`
Definition: specifies connection to stream from. Must be first; one per pipeline. Kafka reads in parallel from source topic partitions; partition limit by processor tier.
Supported source types:
- Kafka broker: `connectionName`, `topic` string/array, optional `timeField`, `partitionIdleTimeout`, `schemaRegistry`, `config` (`auto_offset_reset`, `group_id`, `keyFormat`, `keyFormatError`). If no `timeField`, timestamp comes from source message timestamp.
- MongoDB collection/database/cluster change stream via Atlas connection. Supports `initialSync` for copying existing docs then processing changes; checkpoints initial sync and change phase.
- AWS Kinesis data stream.
- Document array (useful for tests/prototypes).
Late-arriving data for event-time windows can go to DLQ when configured. For Kafka, if a source topic adds partitions while processor runs, ASP continues without reading new partitions; after restart/checkpoint restore processor fails; recreate processors reading topics with newly added partitions.

## Sinks: `$emit` and `$merge`
`$emit`: last stage, one per pipeline. Writes to Kafka, Atlas time series collection, Kinesis, S3, Azure Blob, GCP Pub/Sub. Can route dynamically using field/dynamic expression for target (e.g. Kafka `topic`, Pub/Sub orderingKey/attributes). Expression must resolve to string; failures go to DLQ where supported. Kafka supports schema registry Avro serialization, acks, compression, headers, key/keyFormat, outputFormat basicJson/canonicalJson/relaxedJson, tombstoneWhen. `$emit` to Atlas time series can create collection if DB exists; only one Atlas time series collection per processor. S3 `$emit` cannot write BSON docs >125MB.
`$merge`: last stage, one per pipeline. Writes to Atlas connections only. Modified Data Federation `$merge`: `into` must reference Atlas connection (`connectionName`, `db`, `coll`); `whenMatched` supports `merge`, `replace`, `keepExisting`, `delete`, pipeline/expression and dynamic expressions; `whenNotMatched` supports `insert`, `discard`, `expression`; `parallelism` 1..16 but bounded by tier cumulative parallelism. Dynamic expressions allowed in `into.db` and `into.coll` for per-message routing. `$merge` cannot write time series; use `$emit` for time series. Need Atlas admin role for `$merge` on sharded collections; `on` has special requirements for sharded collections.
`$iceberg`: last stage, one per pipeline, to write Apache Iceberg tables to an AWS S3 bucket through a registered S3 connection. Required fields: `connectionName`, `bucket`, `databaseName`, `tableName`, `path`; `region` required if the processor is not on AWS. `mode:"cdc"` (default) derives insert/update/delete from `stream.source.operationType`; `mode:"insert"` appends rows. Optional `idFieldName` (default `_id`), `partitionedBy` transforms, and `catalog:{type:"hadoop"|"glue"}`. Supports SP10/SP30/SP50 only; dynamic table routing maximum is 5/10/50 tables respectively. It infers/evolves table schema, creates a new table on its first message, uses at-least-once delivery, and sends unsupported BSON types or unresolved dynamic routes to DLQ when configured. Arrays/objects serialize to Basic JSON strings; ObjectId/UUID become strings; dates/timestamps become microsecond UTC timestamptz.

## Windows and timing
Window stages aggregate continuous data over time; stateful, checkpointed, support late-data mechanisms. Apply aggregation queries inside window `pipeline`. One window stage per stream processor.
- Tumbling: non-overlapping contiguous windows with `interval`; optional `offset`; `boundary:"eventTime"|"processingTime"`; optional `idleTimeout`, `allowedLateness` for eventTime.
- Hopping: windows with `interval` and `hopSize`; can overlap; same timing options.
- Session: dynamic activity sessions. Docs with same partition are in same session when timestamp difference < `gap`. Window closes when watermark advances `gap + allowedLateness` beyond max event timestamp in session. Has `partitionBy`, `gap`, `pipeline`, `boundary`, optional allowed lateness/idle handling.
Timing:
- Event time: timestamp from source event/message or `timeField`; late/out-of-order data can miss intended window and go to DLQ.
- Processing time: processor consumption wall-clock time. Cannot specify `idleTimeout` or `allowedLateness` with processingTime windows. Timestamps UTC.
- Watermark: for pipelines with windows; advances only when processor consumes docs with later event time than previously consumed. For Kafka, ASP waits for all partitions to pass watermark; configure partition idle timeout to ignore idle partitions in watermark calculations.

## Enrichment and validation
- `$lookup`: joins streaming docs with Atlas collection from connection registry. Use for enrichment/reference data.
- `$cachedLookup`: same idea with TTL cache and `maxMemUsageBytes` to reduce repeated reads. Does not support `let` or `pipeline`; uses `from`, `localField`, `foreignField`, `as`, `ttl`.
- `$https`: sends HTTPS request per input document via connection; supports dynamic `path`/parameters/body depending syntax; use for external API enrichment. Can be expensive/slow; monitor latency/DLQ.
- `$function`: custom JS function within pipeline. `$externalFunction`: AWS Lambda via external function connection.
- `$validate`: validates docs with query-operator expressions except unsupported `$near`, `$nearSphere`, `$text`, `$where`; `validationAction:"discard"` default or `"dlq"`.

## State, errors, reliability
States include STARTED, STOPPED, PROVISIONING, FAILED and related lifecycle states in APIs/UI. PROVISIONING commonly during startup, scaling, tier/workspace resource allocation; typical startup delay 2-5 minutes.
Error classification:
- User errors: bad credentials, networking/external services, malformed data, permission issues. ASP retries restarts over limited 5-minute period, then FAILED. Some non-retryable fail immediately: `StreamProcessorWorkerOutOfMemory (418)`, `StreamProcessorInvalidOptions (420)`. Manual restart via `start()` after fix.
- Internal errors: retried indefinitely, alert ASP engineering, auto-recover from checkpoint.
Recovery: preserve state via checkpoint, restart, recover from last committed checkpoint, continue processing. If change stream resume token missing from oplog, restart with `resumeFromCheckpoint:false`; reduce risk by setting adequate oplog window.
DLQ: configure Atlas collection in processor definition. `$validate` can send violations to DLQ. DLQ schema documented in monitoring guide; monitor `dlqMessageCount`/size.
At-least-once means duplicate outputs possible after retries/restarts. Design sinks/idempotent writes with deterministic keys/upserts where possible.

## Monitoring, stats, alerts
Atlas UI Monitoring tab per workspace/processor shows runtime stats: messages ingested, processed, sent to DLQ; Kafka source lag by partition and total lag where applicable; filter by processor/time/granularity.
Methods:
- `sp.<name>.sample()` samples current output from running processor.
- `sp.<name>.stats()` returns runtime stats.
Important stats fields: status; inputMessageCount/Size at `$source`; outputMessageCount/Size after full pipeline; dlqMessageCount/Size; changeStreamTimeDifferenceSecs and changeStreamState; latency p50/p99/count/sum over 30s (includes window interval if pipeline has window); stateSize; watermark; operatorStats with per-operator counts/sizes/latency/stateSize/maxMemoryUsage/executionTimeMillis; targetStats for some operators.
State memory limit: `state.stateSize` cannot exceed 80% of pod RAM. Example SP30 8GB => max state 6.4GB. If approaching 80%, stop and restart higher tier or raise workspace max. Crossing threshold fails with Worker out of memory.
Logs: Stream Processing logs available in monitoring; audit logs downloadable. Alerts available for processor/workspace events/metrics.

## Security and access
Roles: Organization Stream Processing Admin, Project Owner, Project Stream Processing Owner, Atlas admin can use ASP. `atlasAdmin` DB role needed to create/run processors via workspace. Org/project access and workspace access controlled in Atlas. Execution profiles control processor permissions/profile. Auditing supported; download audit logs. Networking can use public access lists, VPC peering, Private Link/Private Service Connect depending connection/cloud.

## Limitations
- At-least-once only; duplicates possible.
- No horizontal scaling.
- State size <=80% of RAM; OOM fail when exceeded.
- Pipeline definition <=16MB.
- Usage restricted to required roles.
- Kafka source added partitions not read by running processor; after restart/checkpoint restore fails; recreate processors.
- Kafka protocols/mechanisms limited as listed above.
- Only supported aggregation subset; see docs for exact table.
- Unsupported variables: `$$NOW`, `$$CLUSTER_TIME`, `$$USER_ROLES`, `$SEARCH_META`.
- S3 `$emit` BSON doc limit 125MB.
- `mongosh` >=2.3.4 required to override options on existing processor.
- `$merge` cannot write to time series; `$emit` must be used. `$merge` on sharded collections requires Atlas admin and correct `on` behavior.

## Quick examples
Kafka -> Atlas collection:
```javascript
pipeline = [
{$source: {connectionName: "myKafka", topic: "weather", timeField: {$dateFromString: "$timestamp"}}},
{$match: {"dewPoint.value": {$gt: 5.0}}},
{$merge: {into: {connectionName: "atlasConn", db: "sample_weatherstream", coll: "stream"}}}
]
sp.createStreamProcessor("weatherProc", pipeline, {dlq: {connectionName:"atlasConn", db:"ops", coll:"dlq"}})
sp.weatherProc.start({tier:"SP10"})
```
Kafka -> Kafka dynamic topic:
```javascript
[
{$source: {connectionName:"inKafka", topic:"transactions"}},
{$set: {routeTopic: {$concat:["tenant_", {$toString:"$tenantId"}, "_", "$transactionType"]}}},
{$emit: {connectionName:"outKafka", topic:"$routeTopic", config:{outputFormat:"relaxedJson"}}}
]
```
Tumbling window aggregation:
```javascript
[
{$source: {connectionName:"myKafka", topic:"readings", timeField: {$toDate:"$ts"}}},
{$tumblingWindow: {
boundary:"eventTime",
interval:{size:5, unit:"minute"},
allowedLateness:{size:1, unit:"minute"},
idleTimeout:{size:30, unit:"second"},
pipeline:[{$group:{_id:"$sensorId", avgTemp:{$avg:"$temp"}, count:{$sum:1}}}]
}},
{$merge:{into:{connectionName:"atlasConn", db:"iot", coll:"five_minute_rollups"}}}
]
```

## Changelog highlights through 2026-09-01
- 2026-08-27: Vertical Autoscaling reached GA. Configure optional `minTier` and `maxTier` cost-governance bounds; ASP adjusts a processor's vertical compute capacity based on CPU/memory utilization. It does not add horizontal scaling.
- 2026-08-19: `$externalFunction` gained AWS Lambda Private Link support.
- 2026-08-05: `$externalFunction` gained configurable parallelism.
- 2026-08-04: added Project Stream Processing Manager role, checkpoint-clearing UI when editing processors, new activity-feed events, and Confluent Enterprise Private Link without requiring `dnsDomain`.
- 2026-07-14: `$vectorSearch` added as a GA middle-stage enrichment stage for querying a Vector Search index on a connected Atlas cluster; supports explicit vectors or auto-embedding query text, per-document expressions, optional result pipelines, and `let` variables.
- 2026-07-09: improved Kafka partition distribution under heavy load; Atlas UI Monitoring now shows stream processor CPU and memory usage.
- 2026-06-30: `$iceberg` stage reached GA for writing Apache Iceberg tables to AWS S3.
- 2026-06-23: regional failover support added.
- 2026-06-18: redesigned UI for creating/editing stream processors; in-use connections can be viewed read-only in the UI.
- 2026-06-15: multi-collection `initialSync`; `addedParallelism` in `sp.processor.stats()`.
- 2026-06-11: change stream processor tier through mongosh `sp.<name>.modify()`.
- 2026-06-05: change workspace tier in Atlas UI; visual builder/JSON editor links; connection name special character validation; API 404 fix.
- 2026-05-25: `initialSync` memory handling improvement.
- 2026-05-07: new regions: Azure `brazilsouth`, `australiaeast`; AWS `eu-west-3`; GCP `us-east4`; Azure Blob private networking; display PrivateLink costs; Create Stream Processor API `tier` option.
Always check changelog for features after this date.

## Answering rules for agents
- For “how to build”: workspace -> connections -> pipeline with `$source` first, transforms/enrichment/window (including optional `$vectorSearch` middle-stage enrichment), one sink `$emit` or `$merge` last -> create processor -> start tier -> monitor stats/DLQ.
- For “Kafka CDC/streaming”: mention Kafka connection, topic(s), timeField, partition idle timeout, schema registry if Avro, at-least-once, partition-add limitation.
- For “write to Atlas”: use `$merge` for regular collections/upsert/delete/dynamic db/coll; use `$emit` for time series.
- For “windows”: explain eventTime vs processingTime, watermarks, allowedLateness, idleTimeout, one window per processor.
- For “reliability”: at-least-once + checkpoints + DLQ + idempotent sink design; no exactly-once guarantee.
- For “performance/cost”: choose tier by CPU/RAM/bandwidth/parallelism/Kafka partitions; for variable load, consider GA Vertical Autoscaling with deliberate `minTier`/`maxTier` spend bounds. Monitor stateSize (<80% RAM), CPU/memory, latency, lag, and DLQ. Billing applies while processors run plus transfer/networking.

atlas_stream_processing

Published files and instructions

Redaction report