A hopping window groups stream records into fixed-size time intervals that advance by a shorter interval. A one-minute window advancing every ten seconds produces overlapping results, allowing recent measurements to be compared repeatedly. section 9.1.1
Build a hopping window by configuring three related parts:
- Set the window size with TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)). section 9.1.1
- Set the hop interval with advanceBy(Duration.ofSeconds(10)); an interval shorter than the size creates overlap. section 9.1.1
- Expect windowed aggregation keys to use Windowed, then either unwrap the underlying key or configure a matching time-window Serde. section 9.1.1
IoT temperature aggregation
Group heat-sensor readings by key, aggregate each one-minute window, and emit updated averages every ten seconds. The output can preserve Windowed<String> with WindowedSerdes.timeWindowedSerdeFrom(String.class, 60_000L), or map windowedKey.key() before producing the result. section 9.1.1
Listing 9.3 Adding windowing to the aggregation for the Iot temperature readings
KStream<String,Double> iotHeatSensorStream =
builder.stream("heat-sensor-input",
Consumed.with(stringSerde, doubleSerde));
iotHeatSensorStream.groupByKey() #1
.windowedBy(TimeWindows.ofSizeWithNoGrace(
Duration.ofMinutes(1)) #2
.advanceBy(Duration.ofSeconds(10)))
.aggregate(
() -> new IotSensorAggregation(tempThreshold), #3
aggregator,
Materialized.with(stringSerde, aggregationSerde))
.toStream().to("sensor-agg-output",
Produced.with(serdeString,
sensorAggregationSerde)) #4
Inspect the complete Kafka Streams topology showing grouping, hopping-window configuration, aggregation, materialization, and output. section 9.1.1
TimeWindows align logically to the Unix epoch rather than starting from each event timestamp. Record timestamps determine which aligned window receives each record; they do not determine the window boundaries. section 9.1.5
