Apache Kafka Streams: Building Real-Time Stream Processing Applications

8.2.1 Apache Kafka Streams

Apache Kafka Streams is a robust client library designed for building real-time, scalable, and fault-tolerant stream processing applications within the Kafka ecosystem. It allows developers to process and analyze data stored in Kafka topics with ease, leveraging the distributed nature of Kafka to handle large volumes of data efficiently. In this section, we will explore the core concepts of Kafka Streams, guide you through setting up a Kafka Streams application, and demonstrate its capabilities with practical examples.

Overview of Kafka Streams

Kafka Streams is part of the Apache Kafka ecosystem, providing a lightweight and straightforward approach to stream processing. Unlike other stream processing frameworks, Kafka Streams does not require a separate processing cluster. Instead, it runs as a library within your application, allowing you to deploy stream processing logic alongside your existing services. This integration simplifies operations and reduces the overhead associated with managing separate processing clusters.

Key features of Kafka Streams include:

• Scalability: Kafka Streams can scale horizontally by simply adding more instances of your application.
• Fault Tolerance: Built-in mechanisms ensure data processing continuity even in the face of failures.
• Stateful Processing: Supports complex operations like aggregations and joins with state management.
• Interactive Queries: Allows querying of the state stores for real-time insights.

Setting Up Kafka Streams

Setting up Kafka Streams involves configuring Kafka brokers, topics, and consumer groups. Follow these steps to get started:

1. Install Apache Kafka: Ensure you have Kafka installed and running. You can download it from the Apache Kafka website.

2. Start Kafka Broker and Zookeeper: Use the following commands to start the Kafka broker and Zookeeper:

   # Start Zookeeper
   bin/zookeeper-server-start.sh config/zookeeper.properties
   
   # Start Kafka Broker
   bin/kafka-server-start.sh config/server.properties
   

3. Create Kafka Topics: Use the Kafka CLI to create topics for your streams application:

   bin/kafka-topics.sh --create --topic input-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
   bin/kafka-topics.sh --create --topic output-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
   

4. Configure Kafka Streams Application: Set up your Java application with the necessary dependencies. Add the Kafka Streams library to your pom.xml if you’re using Maven:

   <dependency>
       <groupId>org.apache.kafka</groupId>
       <artifactId>kafka-streams</artifactId>
       <version>3.0.0</version>
   </dependency>
   

5. Define Properties: Configure the Kafka Streams application properties:

   Properties props = new Properties();
   props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-app");
   props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
   props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
   props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
   

Stream Processing Topology

A stream processing topology defines the data flow in a Kafka Streams application. It consists of sources, processors, and sinks that interconnect to process data. Here’s how you can define a simple topology:

StreamsBuilder builder = new StreamsBuilder();

// Define source stream
KStream<String, String> sourceStream = builder.stream("input-topic");

// Define processing logic
KStream<String, String> processedStream = sourceStream
    .filter((key, value) -> value.contains("important"))
    .mapValues(value -> value.toUpperCase());

// Define sink
processedStream.to("output-topic");

// Build topology
KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

In this example, the topology reads data from input-topic, filters messages containing “important”, transforms them to uppercase, and writes the results to output-topic.

Defining Stream Operations

Kafka Streams provides a rich set of operations for processing streams, including filtering, mapping, aggregating, and joining. Let’s explore some of these operations with code examples:

• Filtering: Select messages based on a condition.

  KStream<String, String> filteredStream = sourceStream.filter((key, value) -> value.contains("filter"));
  

• Mapping: Transform each message.

  KStream<String, String> mappedStream = sourceStream.mapValues(value -> "Processed: " + value);
  

• Aggregating: Aggregate messages by key.

  KTable<String, Long> aggregatedTable = sourceStream
      .groupByKey()
      .count(Materialized.as("counts-store"));
  

• Joining: Combine streams based on keys.

  KStream<String, String> otherStream = builder.stream("other-topic");
  KStream<String, String> joinedStream = sourceStream.join(
      otherStream,
      (value1, value2) -> value1 + " joined with " + value2,
      JoinWindows.of(Duration.ofMinutes(5))
  );
  

Stateful Processing

Stateful processing in Kafka Streams is achieved using state stores, which maintain state across messages. This enables operations like windowed aggregations and joins. Here’s an example of a windowed aggregation:

KTable<Windowed<String>, Long> windowedCounts = sourceStream
    .groupByKey()
    .windowedBy(TimeWindows.of(Duration.ofMinutes(1)))
    .count(Materialized.as("windowed-counts-store"));

This example counts messages in one-minute windows, storing the results in a state store.

Fault Tolerance and Recovery

Kafka Streams offers robust fault tolerance through features like automatic failover and state backups. It uses Kafka’s log-based storage to persist state changes, allowing seamless recovery in case of failures. If an instance fails, another instance can take over processing without data loss.

Interactive Queries

Interactive queries allow you to query state stores in real-time, providing insights beyond the stream processing pipeline. You can expose REST endpoints to access the state store data:

ReadOnlyKeyValueStore<String, Long> keyValueStore =
    streams.store(StoreQueryParameters.fromNameAndType("counts-store", QueryableStoreTypes.keyValueStore()));

// Query the store
Long count = keyValueStore.get("some-key");

Example Implementation

Let’s implement a Kafka Streams application that aggregates metrics for a real-time dashboard. We’ll process incoming event data and display aggregated metrics.

1. Setup Kafka Topics:

   bin/kafka-topics.sh --create --topic metrics-input --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
   bin/kafka-topics.sh --create --topic metrics-output --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
   

2. Define the Streams Application:

   StreamsBuilder builder = new StreamsBuilder();
   KStream<String, String> metricsStream = builder.stream("metrics-input");
   
   KTable<String, Long> aggregatedMetrics = metricsStream
       .groupBy((key, value) -> extractMetricKey(value))
       .count(Materialized.as("metrics-store"));
   
   aggregatedMetrics.toStream().to("metrics-output", Produced.with(Serdes.String(), Serdes.Long()));
   
   KafkaStreams streams = new KafkaStreams(builder.build(), props);
   streams.start();
   

   In this example, extractMetricKey is a function that extracts the metric key from the incoming data.

3. Deploy and Monitor:

   Deploy the application and monitor the metrics-output topic for aggregated results. Use tools like Kafka’s command-line utilities or third-party monitoring solutions to visualize the data.

Conclusion

Apache Kafka Streams offers a powerful and flexible framework for building real-time stream processing applications. Its seamless integration with Kafka, combined with features like stateful processing and interactive queries, makes it an ideal choice for developing scalable and fault-tolerant data processing solutions. By following the examples and guidelines provided, you can harness the full potential of Kafka Streams in your event-driven architecture.

Quiz Time!