Data aggregation is a stateful Kafka Streams operation: each incoming record updates a result held for a shared key. Unlike a stateless predicate, which can decide from one event alone, an aggregate retrieves the current value, applies the new event, and stores the updated value. An initializer supplies the first value when no prior state exists. section 7.1
Use aggregate when the result needs a different type than the incoming record. A stream of individual stock transactions, for example, can become a per-symbol summary containing purchase and sale volumes, dollar totals, and the highest and lowest observed prices. This differs from reduce, which returns the same type as its input value. section 7.2.2
A practical stream aggregation follows this sequence:
- Read records into a KStream and group them by the key that identifies the state to maintain, such as a stock symbol. section 7.2.2
- Provide an initializer and an Aggregator that incrementally builds the result for each arriving value. section 7.1 section 7.2.2
- Materialize the result with serdes when the custom aggregate is stored, then convert it to a stream and write updates to an output topic. section 7.2.2
- Choose a result object that holds the running measurements needed by the application, rather than only a count. section 7.2.2
Summarizing stock transactions
Group Transaction records by symbol, initialize an Aggregate, and pass each record to a StockAggregator. The aggregator calculates transaction dollars from shares and price, then updates the running purchase or sale totals and price-related fields. Materializing the Aggregate requires a serde; a Protobuf schema can generate the aggregate type and its serde. The resulting updates can be logged and written to a stock-aggregations topic. section 7.2.2
KTable aggregation has update semantics rather than simple event accumulation. Because the input key acts as a primary key, group records by another field—such as market segment—to combine multiple rows; this regrouping can force repartitioning. Supply an initializer plus two aggregators: an adder for the new value and a subtractor that removes the previous value for an updated key. section 8.4
Figure 8.7 KTable aggregations use an Adder aggregator and a Subtractor aggregator.
Use the adder/subtractor diagram to inspect why a KTable aggregate must remove an old value before incorporating its replacement. section 8.4
An unwindowed aggregate retains all prior history, which can obscure whether a condition is occurring now. For IoT temperature readings, an aggregate can retain the reading count, sum, average, highest temperature, and threshold-exceeded count. Applying windowing separates those results into discrete time blocks, making current spikes easier to identify. section 9.1
Figure 9.8 By using windowing, you can separate the aggregations into discrete blocks of time.
Use the windowing figure to contrast ever-growing aggregate state with time-bounded aggregate results. section 9.1
