KafkaIQ5.1 Stateful Operations
Module Overview
Goal: Explain why state exists in stream processing and how to use stateful operations.
What You'll Learn:
- What makes an operation "stateful"
- Aggregations: count, reduce, aggregate
- GroupBy and GroupByKey
- Why repartitioning happens (conceptually)
- When to use stateful operations
- State vs stateless trade-offs
What's NOT Covered Here:
- Window types (covered in Module 5.2)
- State store internals (covered in Module 6.1)
- Joins (covered in Module 8.1)
- Windowed aggregations (covered in Module 5.3)
Key Insight: Stateful operations require remembering information across multiple records to compute results.
1. What is Stateful Processing?
Definition
A stateful operation needs to remember information from previous records to compute results.
Stateful:
Record 1 → [Process + Store] → State updated
Record 2 → [Process + Store] → State updated (uses Record 1's info)
Record 3 → [Process + Store] → State updated (uses Record 1 + 2's info)
Each record depends on previous records
State maintained in memory/disk
Real-World Analogy: Coffee Shop Register
Imagine you're a barista at a coffee shop with a customer loyalty program:
Stateless Processing = Each Order Processed Independently
Customer Alice orders coffee → Charge $4.50 → Done
Customer Bob orders latte → Charge $5.00 → Done
Customer Alice orders muffin → Charge $3.00 → Done
Each transaction is independent.
You don't remember who ordered what before.
Stateful Processing = Tracking Customer Points
Customer Alice orders coffee ($4.50):
├─ Look up Alice's current points: 10
├─ Add 4 points (1 point per dollar)
└─ Update Alice's state: 14 points
Customer Bob orders latte ($5.00):
├─ Look up Bob's current points: 0
├─ Add 5 points
└─ Update Bob's state: 5 points
Customer Alice orders muffin ($3.00):
├─ Look up Alice's current points: 14
├─ Add 3 points
└─ Update Alice's state: 17 points
You MUST remember each customer's running points total.
Each new order updates their existing state.
The Key Difference:
- Stateless: Process order → Charge → Forget
- Stateful: Process order → Look up state → Update state → Store state
In Kafka Streams, state is stored in state stores (like a database for your streaming app), allowing you to remember information across records.
Spreadsheet Analogy
Think of state like a spreadsheet:
┌────────────┬────────────┬─────────────────┐
│ Customer │ Total │ Orders Count │
│ (Key) │ Spent │ │
├────────────┼────────────┼─────────────────┤
│ Alice │ $7.50 │ 2 │ ← State for Alice
│ Bob │ $5.00 │ 1 │ ← State for Bob
│ Charlie │ $12.00 │ 3 │ ← State for Charlie
└────────────┴────────────┴─────────────────┘
When a new event arrives (Alice orders $3 coffee):
├─ Find Alice's row (key = Alice)
├─ Read current values: $7.50, 2 orders
├─ Update: $7.50 + $3 = $10.50, 2 + 1 = 3
└─ Write back to state store
Each key (customer) has its own row (state).
Stateful operations = reading and writing to this spreadsheet.
Stateful vs Stateless
STATELESS:
├─ filter(value > 100)
├─ mapValues(v -> v.toUpperCase())
└─ Each record processed independently
STATEFUL:
├─ count() → needs to remember previous count
├─ sum() → needs to remember running total
├─ join() → needs to keep records for matching
└─ Requires state storage
Why State is Needed
Question: "How many clicks per user?"
Record 1: user=alice, event=click
Record 2: user=bob, event=click
Record 3: user=alice, event=click
Answer: alice=2, bob=1
To compute this, we must REMEMBER:
├─ How many clicks alice had before (1)
├─ How many clicks bob had before (1)
└─ Update counts as new records arrive
This requires STATE
2. The Aggregation Pattern
Core Concept
Aggregation follows this pattern:
1. Group records by key
2. Apply aggregation function
3. Maintain state (running result)
4. Emit updated result
Example: Word Count
Input Stream:
["hello", "world", "hello", "kafka", "hello"]
Grouped by word:
├─ "hello" → ["hello", "hello", "hello"]
├─ "world" → ["world"]
└─ "kafka" → ["kafka"]
Count per group:
├─ "hello" → 3
├─ "world" → 1
└─ "kafka" → 1
Result: KTable with current counts
3. GroupBy and GroupByKey
groupByKey: Group with Existing Key
KStream<String, Long> input = builder.stream("user-clicks");
// Group by existing key (user ID)
KGroupedStream<String, Long> grouped = input.groupByKey();
Use when: Key already represents what you want to group by.
groupBy: Group with New Key
KStream<String, Order> orders = builder.stream("orders");
// Group by product (not original key)
KGroupedStream<String, Order> byProduct = orders.groupBy(
(orderId, order) -> order.getProductId(),
Grouped.with(Serdes.String(), orderSerde)
);
Use when: Need to group by a field other than the key.
groupBy Triggers Repartitioning
IMPORTANT:
groupBy creates a NEW key
└─ Data must be repartitioned by new key
└─ Writes to internal repartition topic
└─ Reads back in new partition order
└─ Expensive operation
groupByKey keeps existing key
└─ No repartitioning needed
└─ Faster
Prefer groupByKey when possible
Repartitioning covered in detail in Module 7.1.
4. count: Count Records per Key
Syntax
KTable<K, Long> count()
What it does: Count records per key.
Example: Count Events per User
KStream<String, ClickEvent> clicks = builder.stream("clicks");
KTable<String, Long> clickCounts = clicks
.groupByKey()
.count();
// Input:
// user1 → click
// user2 → click
// user1 → click
// user1 → click
// Output (KTable):
// user1 → 3
// user2 → 1
count Creates a KTable
count() returns KTable<K, Long>
Why KTable?
├─ Represents current count per key
├─ Updates as new records arrive
└─ Latest count always available
user1: 0 → 1 → 2 → 3 (as events arrive)
Writing Count Results
// Write to changelog topic
clickCounts.toStream().to("click-counts");
// Or materialize for queries (Module 11.1)
clickCounts.toStream().to("click-counts",
Produced.with(Serdes.String(), Serdes.Long()));
5. reduce: Custom Aggregation Logic
Syntax
KTable<K, V> reduce(Reducer<V> reducer)
What it does: Combine values using custom logic.
Example: Sum Values
KStream<String, Integer> numbers = builder.stream("numbers");
KTable<String, Integer> sums = numbers
.groupByKey()
.reduce((value1, value2) -> value1 + value2);
// Input:
// keyA → 5
// keyA → 3
// keyA → 7
// Output (KTable):
// keyA → 15 (5 + 3 + 7)
Example: Keep Latest Value
KTable<String, Order> latestOrder = orders
.groupByKey()
.reduce((oldOrder, newOrder) -> newOrder); // Keep new
// Equivalent to building a KTable directly
Example: Concatenate Strings
KTable<String, String> concatenated = stream
.groupByKey()
.reduce((str1, str2) -> str1 + "," + str2);
// Input:
// user1 → "a"
// user1 → "b"
// user1 → "c"
// Output:
// user1 → "a,b,c"
6. aggregate: Most Flexible Aggregation
Syntax
KTable<K, VR> aggregate(
Initializer<VR> initializer,
Aggregator<K, V, VR> aggregator
)
What it does: Accumulate values into custom result type.
Example: Count with Custom Type
KStream<String, Order> orders = builder.stream("orders");
KTable<String, OrderStats> stats = orders
.groupByKey()
.aggregate(
// Initializer: create initial value
() -> new OrderStats(0, 0.0),
// Aggregator: update stats with each order
(userId, order, stats) -> {
stats.incrementCount();
stats.addTotal(order.getAmount());
return stats;
},
// Materialized: specify serdes and store name
Materialized.with(Serdes.String(), orderStatsSerde)
);
// Input:
// user1 → Order(amount: 100)
// user1 → Order(amount: 50)
// Output:
// user1 → OrderStats(count: 2, total: 150)
Example: Collect Values into List
KTable<String, List<String>> collected = stream
.groupByKey()
.aggregate(
ArrayList::new, // Initial empty list
(key, value, list) -> {
list.add(value);
return list;
},
Materialized.with(Serdes.String(), listSerde)
);
// Input:
// user1 → "event1"
// user1 → "event2"
// user1 → "event3"
// Output:
// user1 → ["event1", "event2", "event3"]
Why Use aggregate?
Use aggregate when:
✅ Result type differs from input type
✅ Need to maintain complex state
✅ Accumulating into collections
✅ Computing multiple metrics simultaneously
Use reduce when:
✅ Result type same as input type
✅ Simple combination logic
✅ Cleaner syntax
7. Materialized: Naming and Configuring State Stores
What is Materialized?
Controls how state is stored:
Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("click-counts-store")
.withKeySerde(Serdes.String())
.withValueSerde(Serdes.Long())
.withCachingEnabled()
.withLoggingEnabled();
Covered in detail in Module 6.1. For now, basic usage:
KTable<String, Long> counts = stream
.groupByKey()
.count(Materialized.as("my-count-store"));
8. Common Aggregation Patterns
Pattern 1: Count Events per Key
KStream<String, Event> events = builder.stream("events");
KTable<String, Long> counts = events
.groupByKey()
.count();
Use case: Count clicks per user, requests per IP, etc.
Pattern 2: Sum Values per Key
KStream<String, Integer> sales = builder.stream("sales");
KTable<String, Integer> totalSales = sales
.groupByKey()
.reduce((sum1, sum2) -> sum1 + sum2);
Use case: Total sales per product, total bytes per connection.
Pattern 3: Complex Aggregation
KStream<String, Transaction> transactions = builder.stream("transactions");
KTable<String, AccountSummary> summary = transactions
.groupByKey()
.aggregate(
AccountSummary::new,
(accountId, txn, summary) -> {
summary.addTransaction(txn);
summary.updateBalance(txn.getAmount());
summary.recordTimestamp(txn.getTime());
return summary;
},
Materialized.with(Serdes.String(), accountSummarySerde)
);
Use case: Account summaries, session analytics.
Pattern 4: Group by Different Key
KStream<String, Order> orders = builder.stream("orders");
// Group by product instead of order ID
KTable<String, Long> ordersByProduct = orders
.groupBy((orderId, order) -> order.getProductId())
.count();
// Output: count of orders per product
Use case: Group by any field (not just key).
9. State Storage
Where is State Stored?
Aggregation state is stored in:
1. LOCAL STATE STORE (RocksDB or in-memory)
└─ On disk at each Streams instance
2. CHANGELOG TOPIC (Kafka)
└─ Backup in Kafka for fault tolerance
Both automatically managed by Kafka Streams
Covered in detail in Module 6.1.
State Store Implications
STATEFUL OPERATIONS:
1. Use Disk/Memory
└─ RocksDB stores state on disk
2. Slower than Stateless
└─ Read/write to state store
3. Require State Restore
└─ On restart, state rebuilt from changelog
4. Limit Scalability
└─ State tied to partitions
But necessary for aggregations!
10. Performance Characteristics
Aggregation Overhead
Typical latency per record:
Stateless operation: < 1 ms
Stateful operation: 1-10 ms (depends on state size)
Throughput:
Stateless: 1M+ records/sec
Stateful: 100K-500K records/sec
Trade latency for capability
Optimization Tips
1. Use RocksDB (default)
└─ Handles large state efficiently
2. Enable caching
└─ Reduces changelog writes
3. Tune commit interval
└─ Balance durability vs performance
4. Use windowing for bounded state
└─ Covered in Module 5.2
11. Handling Updates and Deletes
Updates
// KTable automatically handles updates
KTable<String, Long> counts = stream.groupByKey().count();
// New record for existing key → updates count
// user1: 1 → 2 → 3 → 4 (as events arrive)
Deletes (Tombstones)
// Null values delete state
KStream<String, String> stream = builder.stream("input");
stream
.groupByKey()
.reduce((v1, v2) -> v2) // Latest value
.toStream()
.filter((k, v) -> v != null) // Filter out deletes
.to("output");
12. Common Pitfalls
Pitfall 1: Using groupBy Unnecessarily
// Bad (unnecessary repartition)
stream.groupBy((k, v) -> k).count()
// Good (no repartition)
stream.groupByKey().count()
Pitfall 2: Large State Per Key
// Bad: unbounded state growth
stream
.groupByKey()
.aggregate(
ArrayList::new,
(k, v, list) -> {
list.add(v); // List grows forever!
return list;
}
);
// Good: use windows (Module 5.2) or bounded collections
Pitfall 3: Wrong Serde for Aggregation Result
// Bad: default serde may not match result type
stream.groupByKey().count().to("output");
// Crashes if default value serde isn't Long
// Good: specify serde
stream.groupByKey()
.count()
.toStream()
.to("output", Produced.with(Serdes.String(), Serdes.Long()));
13. Testing Stateful Operations
Using TopologyTestDriver
// Test aggregation
TopologyTestDriver testDriver = new TopologyTestDriver(topology, props);
TestInputTopic<String, String> input =
testDriver.createInputTopic("input", stringSerializer, stringSerializer);
TestOutputTopic<String, Long> output =
testDriver.createOutputTopic("output", stringDeserializer, longDeserializer);
// Send test data
input.pipeInput("user1", "click");
input.pipeInput("user1", "click");
input.pipeInput("user2", "click");
// Verify results
assertEquals(2L, output.readKeyValue().value); // user1: 2
assertEquals(1L, output.readKeyValue().value); // user2: 1
Testing covered in Module 11.1.
14. Mental Model Summary
Stateful Operations
AGGREGATIONS:
count()
└─ Count records per key
└─ Returns KTable<K, Long>
reduce(reducer)
└─ Combine values (same type)
└─ Returns KTable<K, V>
aggregate(init, aggregator)
└─ Accumulate into custom type
└─ Returns KTable<K, VR>
When to Use Each
Use count() when:
✅ Simple counting
✅ No custom logic needed
Use reduce() when:
✅ Result type = input type
✅ Simple combination (sum, concat, etc.)
Use aggregate() when:
✅ Result type ≠ input type
✅ Complex accumulation logic
✅ Maintaining multiple metrics
The State Lifecycle
1. Group by key
└─ Determine which records belong together
2. Initialize state
└─ Create initial value per key
3. Process each record
└─ Update state with aggregation logic
4. Persist state
└─ Write to local store + changelog
5. Emit results
└─ KTable represents current state
15. Questions You Should Be Able to Answer
After this module, you should be able to answer:
✅ What makes an operation "stateful"?
✅ What's the difference between groupByKey() and groupBy()?
✅ When should I use count(), reduce(), or aggregate()?
✅ What is a KGroupedStream?
✅ Why does groupBy() trigger repartitioning?
✅ Where is aggregation state stored?
✅ What are the performance implications of stateful operations?
16. What's Next?
Now that you understand stateful operations, the next modules will teach you time-based processing:
- Module 5.2: Windowing concepts — time-based grouping
- Module 5.3: Windowed aggregations — combining state with windows
- Module 6.1: State stores — how state is physically stored
- Module 8.1: Joins — another form of stateful processing
Summary
You learned:
- ✅ Stateful operations require maintaining state across multiple records
- ✅ groupByKey() groups by existing key, groupBy() groups by new key (triggers repartition)
- ✅ count() counts records per key
- ✅ reduce() combines values with custom logic (same type)
- ✅ aggregate() accumulates into custom result type
- ✅ State stored in local state store (RocksDB) + changelog topic (Kafka)
- ✅ Stateful operations are slower but necessary for aggregations
- ✅ Results are KTable (current state per key)
Next up: Module 5.2 — Windowing Concepts & Types — Learn how to bound state with time windows.