State stores in Kafka - Mini Project: Build & Apply

Working with State Stores in Kafka Streams

📖 Scenario: You are building a simple Kafka Streams application that counts the number of times each word appears in a stream of text messages. To keep track of the counts, you will use a state store.

🎯 Goal: Create a Kafka Streams application that uses a state store to count word occurrences and prints the counts.

📋 What You'll Learn

Create a Kafka Streams topology with a source topic named input-topic

Create a state store named word-count-store to keep track of word counts

Use a processor to update counts in the state store

Print the word counts from the state store

💡 Why This Matters

🌍 Real World

State stores in Kafka Streams let you keep track of data across messages, like counting words or tracking user sessions.

💼 Career

Understanding state stores is important for building real-time streaming applications that need to remember information between events.

1

Create the Kafka Streams topology with a source

Create a StreamsBuilder object called builder and define a source stream from the topic input-topic using builder.stream("input-topic") assigned to a variable called sourceStream.

Kafka

from kafka import KafkaProducer

# Create a StreamsBuilder and source stream
# Your code here

Need a hint?

Use StreamsBuilder() to create the builder and builder.stream("input-topic") to get the source stream.

2

Add a state store configuration

Create a state store named word-count-store using Stores.keyValueStoreBuilder with Stores.inMemoryKeyValueStore("word-count-store") and Serdes.String() for both key and value serdes. Assign it to a variable called wordCountStore.

Kafka

from kafka.streams import Stores, Serdes

# Create a state store called wordCountStore
# Your code here

Need a hint?

Use Stores.keyValueStoreBuilder with an in-memory store and string serdes for keys and values.

3

Attach the state store and update counts in a processor

Attach the wordCountStore to the topology using builder.addStateStore(wordCountStore). Then, use sourceStream.process() with a processor class called WordCountProcessor that updates counts in the state store. Assume WordCountProcessor is defined to read words and update counts.

Kafka

# Attach the state store and add the processor
# Your code here

Need a hint?

Use builder.addStateStore() to add the store and sourceStream.process() with the processor class and state store list.

4

Print the word counts from the state store

Write code to access the word-count-store from the Kafka Streams context and print all word counts stored. Use store.all() to iterate over all entries and print each word and its count.

Kafka

from kafka.streams.state import ReadOnlyKeyValueStore

# Print all word counts from the state store
# Your code here

Need a hint?

Use store.all() to get all entries and print each key and value.