Atlas Stream Processing processes continuously arriving documents with MongoDB’s document model, Query API, and aggregation pipelines. A stream processor is a pipeline that keeps running: it reads from a source, transforms events, and sends results to a destination, or sink. This supports applications that need to react to live events or combine them with stored data rather than wait for a batch job. section 15.3 section 15.4
A stream processing instance contains processors and a Connection Registry. The registry stores connection settings and credentials for sources and sinks; a connection can be shared by multiple processors, but each processor designates only one source connection and one sink connection. Instances operate independently of Atlas database clusters, so processing workloads can scale without directly scaling those clusters. section 15.4.1 section 15.4.2
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)
Follow events from Kafka or Atlas change streams through continuous processing and time windows to an Atlas database or an external destination. section 15.4.2
Build a processor pipeline in this order:
- Start with the required $source stage to receive documents from a configured connection. section 15.5
- Optionally use $validate to check incoming documents, then apply stages such as $match or $project to process individual documents. section 15.5 section 15.5.3
- For operations over groups of documents, use a time window: $tumblingWindow creates nonoverlapping windows; $hoppingWindow creates overlapping ones. A processor can have only one window stage, and stages such as $group run only inside a window. section 15.5 section 15.5.3
- Send processed results to a destination, such as an Atlas collection or Apache Kafka cluster. section 15.4.2
Inspect a live sample stream
The preconfigured sample_stream_solar connection emits solar-device reports containing observed wattage and temperature. In mongosh, run `let s = {$source: {connectionName: "sample_stream_solar"}}; let processor = [s]; sp.process(processor)`. Documents flow to the console until you press Ctrl+C. This processor is temporary: one created with sp.process() does not persist after termination. section 15.6.3
For operations, checkpoints record processing progress and state so an interrupted processor can resume from its last committed checkpoint. Access can be restricted to individual stream processing instances. External sources such as Kafka also require network setup: allow Atlas IP addresses or use VPC peering. Stream processing instances are supported in specified AWS regions, not on other cloud providers, although processors can interact with clusters in other clouds or regions within the same project. section 15.7.1 section 15.8 section 15.8.3 section 15.4.2
