ATLAS
Atlas Stream Processing
Simplify integrating MongoDB with Apache Kafka to build event-driven applications.
.svg)
A data model built for streaming data
Schema management is critical to data correctness and developer productivity when working with streaming data. The document model gives developers a flexible, natural data model for building apps with real-time data.
A unified developer experience
Developers can use one platform—across API, query language, and data model—to continuously process streaming data from Apache Kafka alongside the critical application data stored in their databases.
Fully managed in Atlas
With a few lines of code, developers can quickly integrate streaming data from Apache Kafka with their database to build reactive, responsive applications—all fully managed with Atlas.
Integrate with Apache Kafka data streams
Atlas Stream Processing makes querying data from Apache Kafka as easy as querying a MongoDB database. A stream processor is made up of a source stage, any number of processing stages, and a sink stage.
xxxxxxxxxx// define a source from the connection registryvar source = { $source: {connectionName: 'kafkaprod',topic: 'stocks'} }// create some other stagesvar match = { $match: { 'exchange':'NYSE'} }// create a sinkvar sink = { $merge: {into: {connectionName: 'mongoprod',db: 'StockDB',col: 'TransactionHistory'} }// try it!var myProcessor = [source, match, sink]sp.process(myProcessor)Perform continuous analytics using window functions
Window operators in Atlas Stream Processing allow you to analyze and process specific, fixed-sized windows of data within a continuous data stream, making it easy to discover patterns and trends in near real-time.
xxxxxxxxxx// define a tumbling window{ $tumblingWindow: { interval: { size: NumberInt(60), unit: 'second'}, pipeline: [{ $group: { _id: '$ip_source', count_connection_reset: { $sum: 1 } } }] } },// output has projections for convenience{ _id: '127.0.0.1', count_connection_reset: 60, _stream_meta: { sourceType: 'kafka', windowStartTimestamp: 2023-05-18T17:07:00.000Z, windowEndTimestamp: 2023-05-18T17:08:00.000Z }}Validate schema on complex events
In Atlas Stream Processing, developers can perform continuous validation. Detecting potential message corruption and late-arriving data ensures that events are properly formed before processing.
xxxxxxxxxx{ $validate: { validator: { $and: [ {$expr: { $ne: [ '$device_id', 'device_8' ] }}, {$jsonSchema: { required: [ 'device_id', 'timestamp', 'obs', 'event_type' ], not : {required : ['event_details']}, properties: { device_id: { bsonType: 'string', pattern: '^device_\\d+', description: "'device_id' is required and must be like device_#" }, obs: { bsonType: 'object', required: [ 'watts', 'temp' ], properties:{ watts : { bsonType : 'int', minimum: 0, maximum: 250,
EVP of Engineering, Acoustic
Learning hub
Find white papers, tutorials, and videos about how to handle streaming data.
_Thumbnail.svg)
_Spot.svg)
_Spot.svg)
FAQ
What is streaming data?
Streaming data is generated continuously from a wide range of sources. IoT sensors, microservices, and mobile devices are all common sources of high-volume streams of data. The continuous nature of streaming data as well as its immutability make it unique from static data at rest in a database.
What is stream processing?
Stream processing is continuously ingesting and transforming event data from an event messaging platform (like Apache Kafka) to perform various functions. This could mean creating simple filters to remove unneeded data, performing aggregations to count or sum data as needed, creating stateful windows, and more. Stream processing can be a differentiating characteristic in event-driven applications, allowing for more reactive, responsive customer experiences.
How is event streaming different from stream processing?
Streaming data lives inside of event streaming platforms (like Apache Kafka), and these systems are essentially an immutable distributed log. Event data is published and consumed from event streaming platforms using APIs.
Developers need to use a stream processor to perform more advanced processing, such as stateful aggregations, window operations, mutations, and creating materialized views. These are similar to the operations one does when running queries on a database, except that stream processing continuously queries an endless stream of data. This area of streaming is an emergent technology with solutions such as Apache Flink and Spark Streaming quickly gaining traction.
With Atlas Stream Processing, MongoDB provides developers with a better way to process streams for use in their applications while leveraging the aggregation framework.
Why did MongoDB build Atlas Stream Processing?
How is stream processing different from batch processing?
Stream processing happens continuously. In the context of building event-driven applications, stream processing enables reactive and compelling experiences like real-time notifications, personalization, route planning, and predictive maintenance.
Batch processing does not work on continuously produced data. Instead, batch processing works by gathering data over a specified period of time and then processing that static data as needed. An example of batch processing is a retail business collecting sales at the close of business each day for reporting purposes or updating inventory levels.
What’s the difference between a stream processing pipeline and an aggregation pipeline?
Atlas Stream Processing extends the aggregation pipeline with stages for processing continuous data streams. These stages combine with existing aggregation stages built into the default mongod process, enabling developers to perform many of the same operations on continuous data as they can perform on data at rest.
Does Atlas Stream Processing support checkpointing?
How does Atlas Stream Processing help me handle data errors during stream processing?
Ready to get started?
- Easily integrate Kafka & MongoDB
- Process data continuously
- Native MongoDB experience
- Available globally