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

A KTable object contains records processed by a Kafka Streams node. If you parallelize your Kafka Streams application by deploying multiple application nodes, then each node contains a table from the records that the node receives. Consequently, the table in each node will not read the records from all partitions.

A GlobalKTable object contains records from all topic partitions. Therefore, if you deploy multiple application nodes, GlobalKTable replicates all records from all partitions on each node. Kafka Streams ensures that the GlobalKTable contains all the records before the stream processing starts.

Stores can be in-memory or persistent, depending on your requirements. A persistent state store allows Kafka Streams to recover transformations from the point where they were left in the case of system failure.

Persistent stores can be saved into the Kafka platform and are fault-tolerant. When you work with the DSL by executing joins and aggregations, Kafka Streams automatically takes care of managing the state as needed.

Sometimes, you might need to interact with the state store directly. For this purpose, Kafka Streams allows you to use interactive queries.

You can query the local state store (the store of every application instance that you are using) by using the DSL. If you want to query all the local state stores of different application instances, then you should add an RPC layer first.

To retrieve data from a store, you should materialize the stream or table. This way, data is retained in the store and you can access it. You can only access the store after the start method has been called on the KafkaStreams class.

The following snippet shows how to retrieve data from a store.

KafkaStreams streams = ...output omitted...;
...output omitted...
streams.start();

ReadOnlyKeyValueStore<Integer, Long> store = streams
            .store(StoreQueryParameters.fromNameAndType( 
                "numbers", 
                QueryableStoreTypes.keyValueStore() 
            ));

store.all()
     .forEachRemaining(row -> {
          System.out.println(row.key + " - " + row.value); 
       });

When aggregations are performed, you can store their result in the state store. Because writing to a persistent storage is expensive, Kafka Streams provides a record cache that you can enable.

With the cache, writes to the persistent storage are prevented by establishing a commit interval. Kafka Streams waits for the commit interval to persist the data.

In this kind of operation, the join is triggered every time that a new element is appended to any of the streams. The window restriction is necessary for efficient joins. An addition to any of the streams triggers a join operation. If you specify no window, then the application must process the full stream every time a new element is appended. Consequently, windowing influences the application performance.

The result is another KStream object. Co-partitioning is required.

For example, consider two streams: visits, which holds the information of visits to a web page, and clicks, which contains the information for clicks in that website. The key for both streams is the user ID. In a specific period of time, you might want to relate the visits and clicks performed by a user. Consider an interval of 5 seconds, the following diagram illustrates the visits and clicks timeline.

Figure 4.1: KStream - KStream join timeline

The letters (A, B, and C) represent the key (the user identifier, in the example) of the records as they arrive to the timeline.

In this kind of operation, the join is triggered every time that a key is updated in any of the tables. Because KTable holds the last value for a specific key, when the join occurs, the last value of every key is used.

The result is another KTable object. No time window is applied. Data co-partitioning is required.

For example, consider two tables: users, which contains the information for every user, and profiles, which contains the public data of every user. Both tables use the user identifier as a key. A KTable is used for both because you are only concerned with the latest update.

Figure 4.2: KTable - KTable join timeline

In the previous operations, the join was triggered every time a change happened in any of the streams (either left or right side). However, when joining a KStream and a KTable, only appends to the KStream (the left side) trigger a join operation. The table (the right operand) just receives updates that are used whenever a new matching key arrives to the stream.

The result is another KStream object. No time window is applied. Data co-partitioning is required.

For example, consider two topics users and orders. Both use the user identifier as a key. The users topic is mapped to a KTable because we just care about the last update of every user. The orders topic is mapped to a KStream because we must consider all orders, not just the last order for every user.

Figure 4.3: KStream - KTable join timeline

This kind of join is similar to the KStream - KTable join. However, the GlobalKTable holds all the values from the topic before the stream processing starts.

The result is another KStream object. No time window is applied. Data co-partitioning is not required.

For example, consider the example from the KStream - KTable join, users and orders. The users topic is now mapped to a GlobalKTable and contains all the values at t=0.

Figure 4.4: KStream - GlobalKTable join timeline

The aggregate operation allows you to update an accumulator object for every element in the grouped stream. An accumulator is a custom object that you provide to the aggregate method and you can update during processing.

The result is a KTable. You can use windowing to aggregate elements in a specific time range.

For example, consider a users stream where you want to find the average age of the users. Conceptually, you would have to sum up all the ages and divide the result by the number of users. With the aggregate operation you have an accumulator object that contains the sum of the ages and the count number of the users. For every element in the grouped stream, you update the accumulator. Consider the following Accumulator class.

class Accumulator {
    int sum = 0;
    int count = 0;

    void addAge(int age) {
        sum += age;
        count++;
    }
}

In the following snippet, the Accumulator class is used to find the age sum.

KStream<Integer, User> usersStream = getUsersStream();

var grouped = usersStream.groupByKey().aggregate( 
            () -> new Accumulator(), 
            (key, user, accumulator) -> accumulator.addAge(user.getAge()), 
            Materialized.<Integer, Accumulator,
                        KeyValueStore<Bytes, byte[]>>as("users-age") 
                    .withKeySerde(intSerde)
                    .withValueSerde(accumulatorSerde)
        );

At the end of the operation, the accumulator contains the sum of the ages and the total number of users processed.

The reduce method is a specific case of the aggregate operation. The reduce method iterates over the elements of the KStream or KTable and for every element, it returns a new element of the same type. The new element is passed to the next iteration, and at the end only one resulting value remains.

The result is a KTable. You can use windowing to reduce elements in a specific time range.

For example, consider a stream of numbers where you want to find the maximum number. For every element in the stream, you verify if the element is greater than the current maximum. If it is, then you return the element, otherwise, you return the current maximum. The following snippet shows a possible implementation.

KStream<String, Integer> stream = ...output omitted...;

stream.groupByKey()
       .reduce(
            (currentMax, element) -> {
                if (element > currentMax) {
                    return element;
                }

                return currentMax;
            }
        );

The count operation allows you to count the number of elements in a grouped KStream or KTable. The result is a KTable. You can use windowing to count elements in a specific time range.

For example, in the previous example of the word grouping, you can count the number of words with the same length.

KStream<String, String> wordsStream = ...output omitted...;

KTable<String, Long> wordsCount = wordsStream
    .groupBy((key, word) -> word.length())
    .count();