Stream Processor Pipeline
Parent: MongoDB Atlas Stream Processing · Published reference · snapshot 2026-10-23 · skill mongodb-cohort-5ceedb05d0f7
↓ Facts as markdownall context files
Source-anchored research for Stream Processor Pipeline. Run status: COMPLETED.
These notes link each claim to its source. Some citations identify archived research reports that are no longer hosted on this site. A published reference means the content is available; it does not certify independent review or accuracy.Read the editorial policy and follow the sources before relying on a claim.
Definitions
- A stream processor is a continuously running MongoDB aggregation pipeline: it ingests streaming documents, transforms them, and sends results to a configured sink. Processing runs separately from the Atlas database cluster. (high confidence) [source]
Structure and components
- The documented processor structure permits one window stage and one sink stage; $merge and $emit are mutually exclusive sink choices. Dynamic routing can select different destinations within the sink connection. (high confidence) (verified as of 2026-10-14) [source]
- With resumeFromCheckpoint=true, source changes, window interval changes, window removal, and window type changes are restricted. A windowed pipeline can be modified only if its inner pipeline contains $group or $sort. (low confidence) (verified as of 2026-10-14) [source]
How-to and procedures
- Use sp.process(pipeline) for interactive prototyping; use sp.createStreamProcessor(name, pipeline) followed by start() for a persistent processor. Sinkless pipelines are supported for ephemeral sp.process() testing. (high confidence) [source]
- Use familiar document transformations such as $match, $project, and $addFields between ingestion and output. A practical pattern filters incoming events, computes or selects fields, then writes the resulting documents to an Atlas collection or a Kafka topic. (high confidence) [source]
- Create a named, persistent processor with sp.createStreamProcessor(name, pipeline, options), then start it explicitly with sp.<name>.start(). Use stop(), stats(), sample(), and listStreamProcessors() to control and inspect it. (high confidence) [source]
- Use sp.process(pipeline, options) for an ephemeral prototype. It runs immediately and does not persist after stopping; a document-array source provides controlled test input before switching to a live stream. (high confidence) [source]
Problems, failure modes and limitations
- Low output is not by itself evidence of failure: selective alert pipelines may emit rarely, while windowed aggregation buffers input. The smart-grid demo groups readings into daily windows before writing aggregate results. (medium confidence) [source]
- Historical issue [redacted] describes a checkpoint-restoration risk: recompiling a pipeline after restart could produce a changed query plan that could not restore prior state. The proposed remedy was to serialize the optimized plan into the checkpoint; the issue is marked Fixed with fix version 8.1.0-rc0. This is historical evidence, not proof of a current service defect. (low confidence) (verified as of 2025-03-11) [source]
- Configure a dead letter queue when defining the processor to retain documents that cannot be processed and their error context in an Atlas collection. Inspect those documents as part of pipeline troubleshooting. (high confidence) [source]
Comparisons and alternatives
- When comparing pipeline engines, verify when results are emitted instead of transferring assumptions: Confluent documents EMIT FINAL to suppress intermediate window results, while MongoDB’s diagnostic reference describes windowed output waiting for window closure. (medium confidence) [source]
- Aggregation syntax is shared with MongoDB, but stage placement is restricted: $group, $sort, $count, and $limit belong inside a window pipeline rather than the main streaming pipeline. (high confidence) (verified as of 2026-10-23) [source]
Facts and statements
- Atlas Stream Processing reuses MongoDB aggregation syntax but supports only a subset of database aggregation stages; a database pipeline is not automatically a valid streaming pipeline. (high confidence) (verified as of 2026-10-14) [source]
- Store a processor as a JSON object with name and pipeline fields. MongoDB’s example repository separates the processor name from the stage array and shows how to load that array for testing or persistent creation. (low confidence) [source]
- For mongosh pipeline changes, stop the processor, call modify(), then start it again. By default, modification restores the last checkpoint and recomputes open windows using the updated pipeline. (low confidence) (verified as of 2026-10-14) [source]
- Do not copy the OneUptime tutorial’s processedAt expression verbatim: its pipeline uses $$NOW, while MongoDB explicitly lists $$NOW as unsupported in Atlas Stream Processing. (high confidence) (verified as of 2026-10-14) [source]
- $source must be the first stage, and each pipeline allows only one $source stage. Kafka messages and Atlas change events are input documents; downstream transformations operate on the document shape supplied by that source. (high confidence) [source]
- Atlas Stream Processing documents only an at-least-once processing guarantee. Checkpoints do not establish exactly-once output; design downstream writes and effects to tolerate duplicates. (low confidence) (verified as of 2026-10-23) [source]
- Do not copy the tutorial expression alertTime: "$$NOW" into a processor: OneUptime uses it in an example, but MongoDB explicitly lists $$NOW among unsupported aggregation variables. (high confidence) (verified as of 2026-10-23) [source]
- Atlas Stream Processing currently uses a single core for transformation stages except where a stage exposes parallelism, and it does not support horizontal scaling. A familiar aggregation pipeline does not imply unrestricted distributed execution. (low confidence) (verified as of 2026-10-23) [source]
Children
- No children recorded.