Developing Event-driven Applications with Apache Kafka and Red Hat AMQ Streams

A processor is a topology node that receives a stream of records as its input. Each processor defines a function that is applied to every record in a stream. Subsequently, Kafka passes processed records to processors further down in the topology.

In a topology, a processor can have an additional function:

Topologies require source and sink processors to feed data into the topology and write the results. Without source and sink processors, topologies are just specifications, without data flowing through the nodes.

A processor can be stateful or stateless. Stateless processors apply functions that do not require a state store, such as the map or filter functions. Stateful processors apply functions that require one or more state stores, such as the join, groupBy, or other aggregation functions. A state store enables a processor to store previous state for further functions, such as interactive querying, or creating data views by using data windows.

Kafka Streams provides a StreamsBuilder object that enables you to define and execute a topology by using a domain-specific language (DSL).

The following simplified Streams application applies the System.out#println function to each record.

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.Topology;
// class omitted

final StreamsBuilder builder = new StreamsBuilder();

builder.stream("input-topic")
  .foreach((key, value) -> System.out.println(value));

final Topology topology = builder.build();

// Property configuration omitted
final KafkaStreams streams = new KafkaStreams(topology, getProperties());
streams.start();

Each additional DSL function creates one or more processors in the application topology. You can query the Topology object by using the Topology#describe function to get a topology description. The previous code sample creates the following topology:

Topologies:
   Sub-topology: 0
    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])
      --> KSTREAM-FOREACH-0000000001

The following is a visualization of the topology description:

Figure 3.3: Stateless Kafka Streams topology

You can also create stateful topologies, for example:

new StreamsBuilder().stream("input-topic")
  .groupBy(
      (key, value) -> value,
      Grouped.with(Serdes.Integer(),Serdes.String())
  ).count();

In the preceding example, the groupBy DSL function selects a new key based on the record value, and filters values with the null key. Then, the count function requires a state store to keep track of the number of records in each category. Consequently, the preceding code generates the following topology:

Topologies:
   Sub-topology: 0
    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])
      --> KSTREAM-KEY-SELECT-0000000001
    Processor: KSTREAM-KEY-SELECT-0000000001 (stores: [])
      --> KSTREAM-FILTER-0000000005
      <-- KSTREAM-SOURCE-0000000000
    Processor: KSTREAM-FILTER-0000000005 (stores: [])
      --> KSTREAM-SINK-0000000004
      <-- KSTREAM-KEY-SELECT-0000000001
    Sink: KSTREAM-SINK-0000000004 (topic: KSTREAM-AGGREGATE-STATE-STORE-0000000002-repartition)
      <-- KSTREAM-FILTER-0000000005

  Sub-topology: 1
    Source: KSTREAM-SOURCE-0000000006 (topics: [KSTREAM-AGGREGATE-STATE-STORE-0000000002-repartition])
      --> KSTREAM-AGGREGATE-0000000003
    Processor: KSTREAM-AGGREGATE-0000000003 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000002])
      --> none
      <-- KSTREAM-SOURCE-0000000006

Figure 3.4: Stateful Kafka Streams topology

See the streams API documentation for further details about the generated topology.

In the preceding stateful example, consider that input-topic contains userID, "link" records. These records log what links a user clicks on a web page:

1, "/technologies"
1, "/our-code-is-open"
2, "/technologies/linux/platform"
5, "/technologies/cloud-computing/openshift"
5, "/technologies"
5, "/our-code-is-open"

The preceding topology counts how many times any user clicked a particular link. Instead of grouping the records by the user ID, the Kafka Streams application groups the records by URL to count records with the same URL value:

"/technologies", 2 // clicked by user IDs 1 and 5
"/our-code-is-open", 2
"/technologies/linux/platform", 1
...

Because the grouping is by URL, and to parallelize the count operation, Kafka Streams sends records with the same URL to the same processing node. Because the Kafka Streams scalability model is based on Kafka partitions, the grouping operation sends such records to the same partition. In the preceding example, Kafka Streams sends records with the value /technologies to the same partition, regardless of the different user ID values.

This operation is called repartitioning. The generated topology first selects records based on the URL values. Then, Kafka assigns the URL values as a key for the selected records. Finally, based on the new key, the records can be repartitioned, which ensures each record with the same URL is counted by the same node with the same local state.

The Kafka Streams DSL provides a number of built-in methods that often correspond to higher-order functions, for example:

Each of the DSL methods adds one or more processors to the topology. If you require greater control over the higher-order functions, then you can implement custom processor classes, and include them in the topology. This is out of the scope for this course.

The stream-table relationship is useful for performing database operations on streams of data. The following operations are common transformations that Kafka Streams enables you to perform by default: