Aggregations

Group records by key and fold them into running totals — count, reduce, aggregate on grouped streams and grouped tables, with the result backed by a state store.

Aggregation in StoatFlow follows the Kafka Streams shape: you first group records by key, then fold them into a running result with count, reduce, or aggregate. The result is a KTable<K, R> — a changelog, one value per key, backed by a state store. This page covers grouping a KStream (groupByKey / groupBy → KGroupedStream) and grouping a KTable (groupBy → KGroupedTable, which adds the adder/subtractor changelog form).

Windowed aggregations (groupByKey().windowedBy(...)) and co-grouping (cogroup) build on the same grouped types but have their own pages — see Windowing. Everything here is non-windowed.

Grouping a stream

Aggregation always starts from a grouped stream. Two ways to get one:

groupByKey() — group by the record's current key. No re-keying, no sub-topology boundary.
groupBy(selector) — derive a new grouping key from each record. This is a key-changing operation: records are re-routed by the new key (StoatFlow does this in-memory between lanes — see Architecture).

Both return a KGroupedStream<K, V>, the entry point to count / reduce / aggregate.

Serdes for the grouping key (and value, where it survives the fold) come from Grouped. Provide them when the engine can't infer them from the source — most importantly after groupBy changes the key type, or when the value serde differs from the stream's default.

import io.stoatflow.core.topology.Grouped
import org.apache.kafka.common.serialization.Serdes

// Group by current key
val byKey = stream.groupByKey(Grouped.with(Serdes.String(), Serdes.String()))

// Re-key, then group (key type changes to the selector's return type)
val byCategory =
    stream.groupBy(
        { _, product -> product.category },
        Grouped.with("by-category", Serdes.String(), productSerde),
    )

import io.stoatflow.core.topology.Grouped;
import org.apache.kafka.common.serialization.Serdes;

// Group by current key
var byKey = stream.groupByKey(Grouped.with(Serdes.String(), Serdes.String()));

// Re-key, then group (key type changes to the selector's return type)
var byCategory =
    stream.groupBy(
        (key, product) -> product.category(),
        Grouped.with("by-category", Serdes.String(), productSerde));

Grouped factories: Grouped.with(keySerde, valueSerde), Grouped.with(name, keySerde, valueSerde), Grouped.as(name), Grouped.keySerde(...), Grouped.valueSerde(...). The name (when given) labels the grouping node in the topology graph and metrics; serdes set here are the default for the downstream aggregation unless Materialized overrides them.

count

count() returns a KTable<K, Long> holding the number of records seen per key. The value serde is Long automatically; you only need a key serde (from Grouped, Materialized, or the topology default). Null values (tombstones) are ignored and do not increment the count.

This is the word-count fold from Your first app, in isolation:

import io.stoatflow.core.state.StateStore
import io.stoatflow.core.topology.Grouped
import io.stoatflow.core.topology.Materialized
import io.stoatflow.core.topology.Named

val counts: KTable<String, Long> =
    words
        .groupBy({ _, word -> word }, Grouped.`as`("group-by-word"))
        .count(
            Named.`as`("count"),
            Materialized.`as`<String, Long, StateStore>("word-counts"),
        )

import io.stoatflow.core.state.StateStore;
import io.stoatflow.core.topology.Grouped;
import io.stoatflow.core.topology.KTable;
import io.stoatflow.core.topology.Materialized;
import io.stoatflow.core.topology.Named;

KTable<String, Long> counts =
    words
        .groupBy((key, word) -> word, Grouped.as("group-by-word"))
        .count(
            Named.as("count"),
            Materialized.<String, Long, StateStore>as("word-counts"));

count() has overloads for (Named), (Materialized), (Named, Materialized), and the no-arg form (auto-named store) — pass only what you need.

reduce

reduce(reducer) folds values of the same type into one per key. The first value for a key is stored as-is; each subsequent value is combined with the stored aggregate via the Reducer<V> — (aggregate, value) -> aggregate. The result is a KTable<K, V>. Null values are ignored.

Use reduce when the running result has the same type as the input (sum, max, latest-wins, string concatenation). The value serde carries through, so it comes from Grouped/Materialized/the default.

import io.stoatflow.core.state.StateStore
import io.stoatflow.core.topology.Materialized
import io.stoatflow.core.topology.Named

// Running total of order amounts per customer
val totals: KTable<String, Long> =
    orderAmounts
        .groupByKey()
        .reduce(
            { runningTotal, amount -> runningTotal + amount },
            Named.`as`("sum-amounts"),
            Materialized.`as`<String, Long, StateStore>("amount-totals"),
        )

import io.stoatflow.core.state.StateStore;
import io.stoatflow.core.topology.KTable;
import io.stoatflow.core.topology.Materialized;
import io.stoatflow.core.topology.Named;

// Running total of order amounts per customer
KTable<String, Long> totals =
    orderAmounts
        .groupByKey()
        .reduce(
            (runningTotal, amount) -> runningTotal + amount,
            Named.as("sum-amounts"),
            Materialized.<String, Long, StateStore>as("amount-totals"));

In Kotlin, reduce takes the reducer lambda first, then optional Named / Materialized. In Java, the reducer is a Reducer<V> — the SAM type for (V, V) -> V — so a lambda passed inline converts to it directly. Overloads accept (Reducer), (Reducer, Named), (Reducer, Materialized), and (Reducer, Named, Materialized).

aggregate

aggregate is the general fold: the result type VR is independent of the input value type. You supply two functions:

Initializer<VR> — () -> VR, the starting aggregate for a key not seen before.
Aggregator<K, V, VR> — (key, value, aggregate) -> aggregate, applied for each record.

The result is a KTable<K, VR>. Because the aggregate type differs from the value type, give it a value serde via Materialized.withValueSerde(...) (or Materialized.with(keySerde, valueSerde)); the key serde still falls back to Grouped/the default.

This mirrors the per-customer daily aggregation in the e-commerce example (examples/ecommerce-daily-customer-behaviour), simplified to a non-windowed running aggregate:

import io.stoatflow.core.state.StateStore
import io.stoatflow.core.topology.Grouped
import io.stoatflow.core.topology.Materialized
import io.stoatflow.core.topology.Named
import org.apache.kafka.common.serialization.Serdes

data class DailyAggregates(
    val productsViewed: Int,
    val purchases: Int,
    val amountSpent: Double,
)

val perCustomer: KTable<String, DailyAggregates> =
    events
        .groupBy(
            { _, e -> e.customerId },
            Grouped.with("by-customer", Serdes.String(), eventSerde),
        )
        .aggregate(
            { DailyAggregates(0, 0, 0.0) },
            { _, event, agg ->
                DailyAggregates(
                    productsViewed = agg.productsViewed + if (event.isView) 1 else 0,
                    purchases = agg.purchases + if (event.isPurchase) 1 else 0,
                    amountSpent = agg.amountSpent + event.amount,
                )
            },
            Named.`as`("daily-aggregation"),
            Materialized.`as`<String, DailyAggregates, StateStore>("daily-aggregation-store")
                .withValueSerde(dailyAggregatesSerde),
        )

import io.stoatflow.core.state.StateStore;
import io.stoatflow.core.topology.Grouped;
import io.stoatflow.core.topology.KTable;
import io.stoatflow.core.topology.Materialized;
import io.stoatflow.core.topology.Named;
import org.apache.kafka.common.serialization.Serdes;

KTable<String, DailyAggregates> perCustomer =
    events
        .groupBy(
            (key, e) -> e.customerId(),
            Grouped.with("by-customer", Serdes.String(), eventSerde))
        .aggregate(
            () -> new DailyAggregates(0, 0, 0.0),
            (key, event, agg) -> new DailyAggregates(
                agg.productsViewed() + (event.isView() ? 1 : 0),
                agg.purchases() + (event.isPurchase() ? 1 : 0),
                agg.amountSpent() + event.amount()),
            Named.as("daily-aggregation"),
            Materialized.<String, DailyAggregates, StateStore>as("daily-aggregation-store")
                .withValueSerde(dailyAggregatesSerde));

The aggregator must be a pure transformation of (key, value, aggregate). Don't mutate the incoming aggregate in place and return it — build and return a new value (the Kotlin example uses a copy; the Java example a new record). The runtime treats the returned object as the new state; mutating shared state defeats the changelog and recovery semantics.

In Kotlin, aggregate also offers a config-first overload — aggregate(named, materialized, initializerFn = { ... }) { k, v, agg -> ... }. Only the final parameter (the aggregator) becomes a trailing lambda block; the initializer is still passed as a regular argument (named here as initializerFn). The Java-friendly overloads take Initializer<VR> and Aggregator<K, V, VR> explicitly.

Aggregating a table: KGroupedTable

KTable.groupBy(selector) produces a KGroupedTable<KR, VR>, not a KGroupedStream. The difference is changelog semantics: a table is a stream of updates, so when a record's grouping key changes (a customer moves department, a product is re-categorised), the aggregate for the old group must be undone and the aggregate for the new group applied. That's why every KGroupedTable fold takes a subtractor alongside the adder.

The source KTable must be materialized (have a state store) for groupBy — the engine reads the old value to know what to subtract. KTable.groupBy will throw at build time otherwise. The selector returns a KeyValue<KR, VR>, so it can change both the grouping key and the value. See KStream & KTable for materialising a table.

count (table)

count() on a KGroupedTable returns KTable<K, Long> and applies the three changelog rules automatically: insert increments, a key change decrements the old key and increments the new, a delete (tombstone) decrements. When a count reaches zero the key is removed from the store.

import io.stoatflow.core.state.KeyValue

// Count users per department, correct under department changes
val usersPerDept: KTable<String, Long> =
    usersTable
        .groupBy { _, user -> KeyValue(user.department, user) }
        .count()

import io.stoatflow.core.state.KeyValue;
import io.stoatflow.core.topology.KTable;

// Count users per department, correct under department changes
KTable<String, Long> usersPerDept =
    usersTable
        .groupBy((key, user) -> KeyValue.pair(user.department(), user))
        .count();

reduce (table)

reduce(adder, subtractor) folds same-typed values. Both are Reducer<V> — (aggregate, value) -> aggregate. On each update the subtractor removes the old value first, then the adder applies the new one.

// Sum scores by department
val scoreByDept: KTable<String, Int> =
    usersTable
        .groupBy { _, user -> KeyValue(user.department, user.score) }
        .reduce(
            adder = { agg, score -> agg + score },
            subtractor = { agg, score -> agg - score },
        )

import io.stoatflow.core.topology.Reducer;

// Sum scores by department
Reducer<Integer> adder = (agg, score) -> agg + score;
Reducer<Integer> subtractor = (agg, score) -> agg - score;

KTable<String, Integer> scoreByDept =
    usersTable
        .groupBy((key, user) -> KeyValue.pair(user.department(), user.score()))
        .reduce(adder, subtractor);

aggregate (table)

aggregate(initializer, adder, subtractor) is the general table fold. The adder and subtractor are both Aggregator<K, V, VR> — (key, value, aggregate) -> aggregate (Kafka Streams uses Aggregator for both roles; there is no separate Subtractor type). Subtractor runs first on an update, then adder.

// Track the set of user IDs per department
val membersByDept: KTable<String, Set<String>> =
    usersTable
        .groupBy { _, user -> KeyValue(user.department, user) }
        .aggregate(
            initializer = { mutableSetOf<String>() },
            adder = { _, user, set -> set.also { it.add(user.id) } },
            subtractor = { _, user, set -> set.also { it.remove(user.id) } },
        )

import io.stoatflow.core.topology.Aggregator;
import io.stoatflow.core.topology.Initializer;

import java.util.HashSet;
import java.util.Set;

// Track the set of user IDs per department
Initializer<Set<String>> initializer = HashSet::new;
Aggregator<String, User, Set<String>> adder = (dept, user, set) -> {
    set.add(user.id());
    return set;
};
Aggregator<String, User, Set<String>> subtractor = (dept, user, set) -> {
    set.remove(user.id());
    return set;
};

KTable<String, Set<String>> membersByDept =
    usersTable
        .groupBy((key, user) -> KeyValue.pair(user.department(), user))
        .aggregate(initializer, adder, subtractor);

When the source table is backed by a versioned store, out-of-order records (timestamp older than the last seen for that key) are ignored and do not update the aggregate — this prevents late records from corrupting the result. See State stores.

The functional interfaces

These are the Java SAM types used across all aggregation methods (Kotlin callers pass lambdas directly). They are import-compatible in name and shape with their Kafka Streams equivalents.

Interface	Signature	Used by
`Initializer<VA>`	`() -> VA`	every `aggregate` (initial value)
`Aggregator<K, V, VA>`	`(key, value, aggregate) -> aggregate`	every `aggregate` — adder and the `KGroupedTable.aggregate` subtractor (KS uses `Aggregator` for both)
`Reducer<V>`	`(aggregate, value) -> aggregate`	every `reduce` (and table adder/subtractor)

All live in io.stoatflow.core.topology. For the full DSL parity status against Kafka Streams, see the Kafka Streams compatibility matrix.

Materializing the result

Every aggregation writes to a state store. You control it with Materialized:

Materialized.as("store-name") — name the store explicitly (recommended; the name is the store's stable identity for changelog topics and recovery).
.withValueSerde(serde) — required when the aggregate type differs from the input value type (i.e. for aggregate).
Materialized.with(keySerde, valueSerde) — set both serdes at once.
Materialized.valueSerde(serde) — value serde only, auto-named store.

If you omit Materialized, the store gets an auto-generated name and falls back to Grouped / the topology default serdes. Stores are durable by default — every aggregation update is committed to a Kafka changelog topic on the commit barrier, so the running result survives restarts. The full store configuration surface (RocksDB vs in-memory, logging, caching, retention) is on State stores.

Consuming the result

count / reduce / aggregate all return a KTable. To emit the running result to a topic, turn it back into a changelog stream with toStream() and to(...):

import io.stoatflow.core.topology.Named
import io.stoatflow.core.topology.Produced
import org.apache.kafka.common.serialization.Serdes

counts
    .toStream(Named.`as`("to-stream"))
    .to(
        "word-counts",
        Produced.`as`<String, Long>("sink")
            .withKeySerde(Serdes.String())
            .withValueSerde(Serdes.Long()),
    )

import io.stoatflow.core.topology.Named;
import io.stoatflow.core.topology.Produced;
import org.apache.kafka.common.serialization.Serdes;

counts
    .toStream(Named.as("to-stream"))
    .to(
        "word-counts",
        Produced.<String, Long>as("sink")
            .withKeySerde(Serdes.String())
            .withValueSerde(Serdes.Long()));

Because the output is a changelog, downstream consumers see every intermediate update of a key's aggregate, not just a final value — that's the expected KTable semantics, the same as Kafka Streams.

Next steps

Windowing — time, sliding, and session windows on grouped streams, plus cogroup.
KStream & KTable — materialising a table so KTable.groupBy can track old values.
State stores — store types, changelog, in-memory vs RocksDB, versioned stores.
Testing — assert on aggregation results with the in-memory test driver.

KStream and KTable operations

Stateless stream transforms (map, filter, flatMap, selectKey, branch, merge) and KTable basics — with the rules for when a key change forces records onto a different lane.

Windowing

Windowed aggregations in StoatFlow — tumbling, hopping, sliding, and session windows; emit strategies, suppression, Windowed keys, and WindowedSerdes.