An Aggregator is a Kafka Streams functional interface whose implementation updates an aggregate result from each input record and the current aggregate value. It is useful when a simple reduction cannot represent the required result, such as summarizing stock transactions into purchase volume, sale volume, dollar volume, and price extrema. section 7.2.2
A Kafka Streams aggregation connects several pieces:
- The input record supplies the key and value; the current aggregate supplies the accumulated state. section 7.2.2
- An initializer creates the default aggregate for the first record. section 8.4
- A concrete Aggregator implementation can return a different type from the incoming value and can be tested independently. section 7.2.2
- Materialization stores the aggregate with an appropriate Serde, while toStream and to publish the results. section 7.2.2
Summarizing stock transactions
A stream reads Transaction records from stock-transactions, groups them by symbol, and calls aggregate(() -> initialAggregate, new StockAggregator(), Materialized.with(...)). StockAggregator applies each transaction to the current Aggregate, calculating transaction dollars from share count and price, then updates purchase or sale totals. The resulting KTable is converted to a stream and written to stock-aggregations. section 7.2.2
Listing 7.4 Aggregator implementation used for creating stock transaction summaries
public class StockAggregator implements Aggregator<String,
Transaction,
Aggregate> {
@Override
public Aggregate apply(String key,
Transaction transaction,
Aggregate aggregate) { #1
Aggregate.Builder currAggregate =
aggregate.toBuilder(); #2
double transactionDollars =
transaction.getNumberShares()
* transaction.getSharePrice(); #3
if (transaction.getIsPurchase()) { #4
long currentPurchaseVolume =
currAggregate.getPurchaseShareVolume();
currAggregate.setPurchaseShareVolume(
currentPurchaseVolume
+ transaction.getNumberShares());
double currentPurchaseDollars =
currAggregate.getPurchaseDollarAmount();
currAggregate.setPurchaseDollarAmount(
currentPurchaseDollars
+ transactionDollars);
}
Inspect the implementation pattern: apply receives the key, Transaction, and current Aggregate; the method copies the aggregate, calculates transaction value, and updates the relevant fields. section 7.2.2
Listing 7.5 Kafka Streams aggregation
KStream<String, Transaction> transactionKStream =
builder.stream("stock-transactions",
Consumed.with(stringSerde, txnSerde)); #1
transactionKStream.groupBy((key, value) -> value.getSymbol(), #2
Grouped.with(Serdes.String(), txnSerde))
.aggregate(() -> initialAggregate, #3
new StockAggregator(),
Materialized.with(stringSerde, aggregateSerde))
.toStream() #4
.peek((key, value) -> LOG.info("Aggregation result {}", value))
.to("stock-aggregations", Produced.with(stringSerde, aggregateSerde)); #5
Follow the complete topology from source topic and grouping through initialization, aggregation, materialization, stream conversion, logging, and output topic. section 7.2.2
KTable aggregation adds a second Aggregator: one adds the new value, and another subtracts the previous value associated with the same table key. This maintains grouped results as table entries change. section 8.4
