A time window divides a continuous stream into time-delimited sets of records. This gives stateful operations a bounded set to work on, so a stream processor can calculate counts, sums, or averages instead of trying to aggregate an infinite stream. section 15.2.2 section 15.5
Choose the window shape to match the result you need:
- Tumbling windows are fixed and nonoverlapping. Each event belongs to one window, making them suitable for distinct totals such as hourly sales or daily reports. In Atlas Stream Processing, use $tumblingWindow with a duration. section 15.2.2 section 15.5
- Hopping windows can overlap. A window has a duration and a hop interval, so one event may contribute to several results. Use $hoppingWindow for rolling measures such as a moving average or continuous anomaly monitoring. section 15.2.2 section 15.5
Place one window stage in a stream processor to define the sets used by stateful stages such as $group and $avg. After processing, choose one output: $emit sends results to a streaming platform, while $merge writes them to an Atlas database. Window choice affects both the accuracy and performance of the analysis. section 15.5 section 15.2.2
Rolling sensor temperature
Read IoT sensor readings from a Kafka topic, start a 30-second hopping window every 10 seconds, and calculate the average temperature within each window. Write the results to a MongoDB collection for further analysis. Because the windows overlap, a reading can affect more than one average. section 15.6.4 section 15.2.2
Listing 15.6 Stream processor with $hoppingWindow for IoT sensor data
// Initially, the processor is empty.
let processor = [] // #1
// The source retrieves data from a Kafka topic named "sensorData".
let source = {
$source: {
connectionName: "iotSensorConnection",
➥// placeholder Kafka connection for IoT data
topic: "sensorData" // Kafka topic with sensor data
}
} // #2
// Add the source to the processor.
processor = [source] // #3
// Use $hoppingWindow to create overlapping time windows of
➥30 seconds, starting every 10 seconds.
let hoppingWindow = {
$hoppingWindow: {
interval: {
size: 30, // Each window lasts 30 seconds
unit: "second"
},
hopSize: {
size: 10, // Windows start every 10 seconds
unit: "second"
},
pipeline: [
{
$group: {
_id: "$_id", // Use the default _id field for the window
avgTemperature: { $avg: "$temperature" }
➥// Calculate the average temperature from sensor data
}
}
]
}
} // #4
// Add the hopping window stage to the processor.
processor.push(hoppingWindow) // #5
// The sink is the final stage. It specifies where
➥the aggregated data should go.
let sink = {
$merge: {
into: {
connectionName: "processedSensorData",
➥// placeholder connection name for storing processed data
db: "iotData",
➥// MongoDB database for sensor data
coll: "temperatureAverages"
➥// Collection for storing average temperatures
}
}
} // #6
// Add the sink to the processor.
processor.push(sink) // #7
// The complete processor pulls data from the Kafka topic,
➥applies the hopping window,
// calculates average temperatures, and stores the results
➥in a MongoDB collection.
sp.createStreamProcessor(
➥'iotTemperatureProcessor', processor) // #8
// Start the processor, which will run continuously.
sp.iotTemperatureProcessor.start() // #9
The listing shows the Kafka source and the $hoppingWindow interval, hop size, and aggregation pipeline for the sensor example. section 15.6.4
