A stream processor is a MongoDB aggregation pipeline that runs continuously. It consumes documents from a source, processes them as they arrive, and sends results to a destination. Unlike a regular query, it does not finish after returning one result set. This makes it useful for keeping a collection or downstream stream current without repeatedly running batch jobs. section 15.4 section 15.4.2
Build the pipeline as an ordered array of stages:
- Start with exactly one $source. Its connection can supply data from Apache Kafka, a MongoDB change stream, or a document array. section 15.9
- Add processing stages as needed. For example, $match filters documents, $validate checks their schema, and $lookup enriches them from an Atlas collection. section 15.5.3 section 15.6.4
- Bound streaming aggregations with a window: $tumblingWindow makes consecutive, nonoverlapping intervals; $hoppingWindow makes overlapping intervals. Put stages such as $group inside the window pipeline. section 15.5.3 section 15.6.6
- Finish with an output stage when results must be retained or forwarded. $merge writes to an Atlas collection; $emit sends processed messages to a configured stream or time-series collection. section 15.6.4 section 15.5.3
Preview a live source
In mongosh, try `let processor = [{$source: {connectionName: "sample_stream_solar"}}]; sp.process(processor)`. The sample connection continuously produces solar-device reports, and this temporary processor prints the flowing documents in the shell. Press Ctrl+C to stop it; a processor created with sp.process() does not persist after termination. For a lasting result, configure the destination connection, add a window and $merge, then create and start a named processor with sp.createStreamProcessor() and its start() method. section 15.6.3 section 15.6.6
Connections and processors belong to an Atlas stream processing instance. Its Connection Registry stores endpoints and credentials; connection names in illustrative pipelines are placeholders until you configure them. Inspect a running processor with stats() for status, message counts, and memory use, or sample() for processed documents. Checkpoints preserve processing position and state so an interrupted processor can resume from its last committed checkpoint. Stream processing workloads scale independently of Atlas clusters, but the cited deployment support is limited to AWS regions; pipelines are limited to 16 MB. section 15.4.1 section 15.6.4 section 15.6.6 section 15.7.1
Figure 15.5 The process flow in Atlas Stream Processing. On the left, data from Apache Kafka is ingested into Atlas through Kafka or via Atlas change streams, which capture real-time data changes. This data is passed to Atlas Stream Processing, where it is processed continuously via operations such as aggregation, filtering, and routing. The figure highlights the stateful windows used within the processing to manage and analyze data over specific time intervals. After processing, the results are materialized into an Atlas database for storage, or the data can be sent back to external systems like Apache Kafka for further use. (Image © MongoDB 2024 CC BY-NC-SA 3.0)
Trace the full flow: Kafka or Atlas change streams supply events, processing stages and stateful windows transform them, and results go to Atlas or Kafka. section 15.4.2
