A hopping window divides a data stream into time-limited windows that can overlap. Its duration sets how much time each window covers; its hop interval sets how often a new window starts. One event can therefore contribute to several results. Use this pattern for rolling averages or monitoring that needs frequent updates. section 15.2.2 section 15.5
In Atlas Stream Processing, $hoppingWindow is a pipeline stage. Windowing makes an unbounded stream available to stateful operations such as $group and $avg. A stream processor can contain only one window stage. Choose $tumblingWindow instead when intervals must not overlap and each event should belong to just one window. section 15.5 section 15.2.2
To calculate rolling sensor temperatures:
- Read sensor events from a Kafka topic with $source. section 15.6.4
- Set $hoppingWindow.interval to 30 seconds and hopSize to 10 seconds, then calculate an average temperature within each window. section 15.6.4
- Write the aggregated results to a MongoDB collection for later analysis. section 15.6.4
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 IoT processor listing shows the $hoppingWindow configuration in its source-to-output pipeline. section 15.6.4
