KGroupedTable is the intermediate table returned by KTable.groupBy. It changes the grouping key so aggregate operations can combine multiple primary-key records, often requiring repartitioning. section 8.4
A grouped-table aggregation typically proceeds as follows:
- Group each table record by another field, such as market segment, rather than its original primary key. section 8.4
- Provide an Initializer for the first aggregate value. section 8.4
- Provide an adder and subtractor Aggregator so updates add the new value and remove the previous value. section 8.4
Market-segment stock totals
A stock-alert KTable can group records by market segment, aggregate share volume and dollar volume, materialize the result, and publish it as a stream to an output topic. section 8.4
Listing 8.2 Aggregates with a KTable
KTable<String, StockAlert> stockTable =
builder.table("stock-alert",
Consumed.with(stringSerde, stockAlertSerde)); #1
stockTable.groupBy((key, value) ->
KeyValue.pair(value.getMarketSegment(), value),
Grouped.with(stringSerde, stockAlertSerde)) #2
.aggregate(segmentInitializer, #3
adderAggregator, #4
subtractorAggregator, #5
Materialized.with(stringSerde, segmentSerde))
.toStream()
.to("stock-alert-aggregate",
Produced.with(stringSerde, segmentSerde));
Follow the complete KTable pipeline from source table and groupBy through aggregation, materialization, and output. section 8.4
