We will use some API methods on this instance in the next code block. But first, lets set up our REST service to start listening on the appropriate port (8080 in this example). As we recently learned, we can use a point lookup to retrieve a single value from our state store. Furthermore, instead of returning a KGroupedStream, invoking groupBy on a KTable returns a different intermediate representation: KGroupedTable. So, in this case, we can use a local state, since a particular instance do not need to share with others, because it has a particular partition, exclusive to it. The concept of cluster do not apply to it. An example response to an interactive query is shown in the following code block: In this chapter, you learned how Kafka Streams captures information about the events it consumes, and how to leverage the remembered information (state) to perform more advanced stream processing tasks, including: Rekeying messages to meet the co-partitioning requirements for certain join types, Grouping records into intermediate representations (KGroupedStream, KGroupedTable) to prepare our data for aggregating, Using the interactive queries to expose the state of our application using both local and remote queries. The most common method for combining datasets in the relational world is through joins.12 In relational systems, data is often highly dimensional and scattered across many different tables. To execute this type of query against a simple key-value store, we could run the A Stream is an infinite sequence of messages. The keys are also used as join keys, to relate both streams. Since grouping data is a prerequisite for aggregating, we need to group the enriched stream. Without further ado, lets take a look at the architecture of our video game leaderboard. We can now run our application, generate some dummy data, and query our leaderboard These point-in-time representations, or snapshots, are referred to as tables, and Kafka Streams includes different types of table abstractions that well learn about in this chapter. In order to do this, we need to materialize the state store. There are multiple state stores supported, including: QueryableStoreTypes.timestampedKeyValueStore(), QueryableStoreTypes.timestampedWindowStore(). In addition to being smaller, the data in the products topic is also relatively static. That means: We can print whatever you need on a massive variety of mediums. However, when building microservices using Kafka Streams interactive queries feature, which we will discuss later in Interactive Queries, clients require only read access to the underlying state. Branch is a very interesting concept. A changelog topic for backing the state store, which is used by the join operator. Everything is interconnected, and by capturing and remembering facts, we can begin to understand their meaning. It is common to see these same patterns in Kafka as well, either because events are sourced from multiple locations, developers are comfortable or used to relational data models, or because certain Kafka integrations (e.g., the JDBC Kafka Connector, Debezium, Maxwell, etc.) The semantics are a little different across each, since streams are immutable while tables are mutable. For example, what if you need to count the number of entries across all of your distributed state stores? You see, events (or facts) rarely occur in isolation in the real world. We could also simply use a static method reference here, such as So, the scenarios of duplicate writer and duplicate processing do not occur. However, we still have one final step to tackle in order to expose the leaderboard results to external clients. Now, lets implement the /leaderboard/:key endpoint, which will show the high scores for a given key (which in this case is a product ID). Aggregate the grouped stream. If we produce this dummy data into the appropriate topics and then query our leaderboard service, we will see that our Kafka Streams application not only processed the high scores, but is now exposing the results of our stateful operations. The video game industry is a prime example of where stream processing excels, since both gamers and game systems require low-latency processing and immediate feedback. Enrich an event with additional information or context that was captured in a separate stream or table, Compute a continuously updating mathematical or combinatorial transformation of related events, Group events that have close temporal proximity. We have already established that stateful operations require our application to maintain some memory of previously seen events. This is possible by understanding events in their larger historical context, or by looking at other, related events that have been captured and stored by our application. For stream-stream and table-table joins: same semantics as a stream-stream left join, except an input on the right side of the join can also trigger a lookup. Each operator is detailed in Table4-4. We can use a lambda to select the new key, since the groupBy operator expects a KeyValueMapper, which happens to be a functional interface. The good practice here is just to use a new consumer group for this new app, to start reading from the first offset of the input topic, generating a new output topic. 13 If youre not using Confluent Platform, the script is kafka-topics.sh. Read the players topic into a KTable since the keyspace is large (allowing us to shard the state across multiple application instances) and since we want time synchronized processing for the score-events -> players join. Another benefit of stateful stream processing is that it gives us an additional abstraction for representing data. For this reason, we will spend some time learning about the inner workings of stateful processing in Kafka Streams before we start using the stateful operators listed in Table4-1. State stores support multiple access modes and query patterns. For the keyed topics (players and products), the record key is formatted as For example, performing a windowed join allows us to understand how discrete event streams relate during a certain period of time. Kafka Streams always writes aggreegation result in log-compacted result topics ( only the latest state is preserved). 10 We can also use KStreams for lookup/join operations, but this is always a windowed operation, so we have reserved discussion of this topic until the next chapter. In this tutorial, our score-events topic contains raw score events, which are unkeyed (and therefore, distributed in a round-robin fashion) in an uncompacted topic. Internet of Things: In stream processing, it is a common scenario to try to predict when maintenance is required, by reading sensor information from devices. In the previous chapter, we learned how to perform stateless transformations of record streams using the KStream abstraction and a rich set of stateless operators that are available in Kafka Streams. Specifically, well be looking at how to use Kafka Streams table abstractions to model data as a sequence of updates. Operations that leverage state stores are key-based. The records are unkeyed and are therefore distributed in a round-robin fashion across the topics partitions. Regarding the last two points, you have a lot of flexibility in choosing which server and client components you want to use for inter-instance communication. Since Kafka Streams inherits Kafka characteristics, it also can be considered fault-tolerant and scalable. To complicate things slightly, Kafka Streams explicitly refers to certain types of state stores as key-value stores, even though all of the default state stores are key-based. It is guaranteed that no input message will be missed or no duplication will be produced. Lets consider that we have to reach a database in order to process our stream. In order to query the full state of our application, we need to: Discover which instances contain the various fragments of our application state, Add a remote procedure call (RPC) or REST service to expose the local state to other running application instances18, Add an RPC or REST client for querying remote state stores from a running application instance. We will accomplish this by building a RESTful microservice using the interactive queries feature in Kafka Streams. The following code block shows a simple implementation of the leaderboard service: HostInfo is a simple wrapper class in Kafka Streams that contains a hostname and port. Streams and tables are the same, but represented differently. The products topic is similar to the players topic in terms of configuration (its compacted) and its bounded keyspace (we maintain the latest record for each unique product ID, and there are only a fixed number of products that we track). For now, its sufficient to understand that co-partitioning is simply an extra set of requirements that are needed to actually perform the join. Instead, we would leverage another Kafka Streams method, called allMetadataForStore, which returns the endpoint for every running Kafka Streams application that shares the same application ID and has at least one active partition for the provided store name. Our updated abstraction table now looks like this: We have one topic left: the products topic. Now that we understand the benefits of stateful stream processing, and the differences between facts and behaviors, lets get a preview of the stateful operators in Kafka Streams. However, collecting, remembering, and analyzing each of the facts (which is what stateful processing enables) allows us to recognize and react to the behavior, and provides much greater business value than viewing the world as a series of unrelated events. Each record is keyed by a product ID. Whenever you use a stateful operator in your Kafka Streams application, its helpful to consider which type of state store is needed by the operator, and also how to configure the state store based on your optimization criteria (e.g., are you optimizing for high throughput, operational simplicity, fast recovery times in the event of failure, etc.). This will be communicated to other running application instances through Kafkas consumer group protocol. 9 As mentioned in Chapter3, if our topics contained Avro data, we could define our data model in an Avro schema file instead. For instance, if your requirement asks you to calculate the min and max value of a specific time window, you need to save state somehow, and you can adopt shared state or local state in order to accomplish it. More details related to Exactly Once capabilities can be found in the three articles below: Some intermediate topics are created internally by Kafka Streams: Messages in those topics will be saved in a compacted way. Windows can be aligned to a clock time (for instance, a 5 minute interval that moves every minute will have the first slice at 00:00-00:05 and the second slice at 01:00-06:00) or can be unaligned (and will start when application starts; the first slice can be 12:12-12:17 and the second slice will be 12:13-12:18). A Kafka Streams application is just a standard java application (jar) that runs in a JVM as an isolated process. About the When: assignation occurs at rebalancing, which means when adding or removing a consumer/partition. We will look at aggregations shortly, but first, lets take a look at how to group KTables. Similar to a range scan, the all() query returns an iterator of key-value pairs, and is similar to an unfiltered SELECT * query. The storage abstraction that addresses these needs in Kafka Streams is called a state store, and since a single Kafka Streams application can leverage many stateful operators, a single application may contain several state stores. Instead, it is expected (as near real time) to have, in minutes, the confirmation sent by email, the credit card charged on time, and additional information of customer history appended to the reservation, for additional analysis. OK, were ready to move on to the second join. View all OReilly videos, Superstream events, and Meet the Expert sessions on your home TV. Fortunately, Kafka Streams provides a method called queryMetadataForKey,19 which allows us to discover the application instance (local or remote) that a specific key lives on. First, lets start with a simple wrapper class for the score-events -> players join. Join the score-events stream and the players table. Each event is treated as an independent and atomic fact, which can be Review of the streams concepts in Apache Kafka. We will discuss these commonalities in this section to get a better idea of how state stores work. The interface that helps us with this is Initializer, and like many of the classes in the Kafka Streams API, Initializer is a functional interface (i.e., contains a single method), and therefore can be defined as a lambda. But what if you need to execute a query that doesnt operate on a single key? Grouped allows us to pass in some additional options for grouping, including the key and value Serdes to use when serializing the records. A Source Stream Processor is the one who connects to a topic to bring data to the topology. 16 We havent talked about deleting keys yet, but we will cover this topic in Chapter6, when we discuss cleaning up state stores. Persistent state stores flush state to disk asynchronously (to a configurable state directory), which has two primary benefits: State can exceed the size of available memory. We will also use OkHttp, developed by Square, for our REST client for its ease of use. So, lets say that the expected time from search to click is five seconds. Stream Processor is a node on that Topology, responsible to process the stream in a specific way. Stream Processors are connected by Streams. Gets one record and returns zero, one or more than one records (an iterable). Once weve created our real-time leaderboard using a new set of stateful operators, we will demonstrate how to query Kafka Streams for the latest leaderboard information using interactive queries. The two topics must be joined considering a time window. By partitioning the state in this way, we can lower the local storage overhead for each individual Kafka Streams instance. By sending data to a log compacted topic, Kafka will make sure that this will occurr as expected. For example, if you were to look at the internals of the count aggregation, youd see an initializer that sets the initial value of the aggregation to 0: The initializer is defined as a lambda, since the Initializer interface is a functional interface. As shown in Table4-5, co-partitioning is not required for GlobalKTable joins since the state is fully replicated across each instance of our Kafka Streams app. Since we only want to trigger the join when the ScoreEvent record can be matched to a Player record (using the record key), well perform an inner join using the join operator, as shown here: The join parameters define how the keys and values for the join records should be serialized. The next thing we need to tackle is grouping the enriched records so that we can perform an aggregation. We have now completed step 5 of our leaderboard topology (Figure4-1). A Join mechanism relates a KTable/KStream with another, creating another stream as part of it. All of the default state stores leverage RocksDB under the hood. As always, well start by defining our data models. Regarding the performance gains of an in-memory state store, these may not be drastic enough to warrant their use (since failure recovery takes longer). RocksDB is a fast, embedded key-value store that was originally developed at Facebook. 14 The GlobalKTable side of the join will still use the record key for the lookup. The difference lies in the return type. the userId is not relevant anymore; grouping is being performed over the last color choose by each user; only colors should be counted, not users. It triggers repartition, because key changes as a result of an aggregation. When we need to access a state store in read-only mode, we need two pieces of We will use an inner join, using the join operator for each of the joins since we only want the join to be triggered when theres a match on both sides. Since this is the exactly situation of a KTable, it is right to assume that each topic that uses KTable would be set as log-compacted. Materialized state stores differ from internal state stores in that they are explicitly named and are queryable outside of the processor topology. Figure: Multiphase Processing - from Kafka: The Definitive Guide e-book. The first thing we need to do when adding a source processor is determine which Kafka Streams abstraction we should use for representing the data in the underlying topic. When performing a join using traditional SQL, we simply need to use the join operator in conjunction with a SELECT clause to specify the shape (or projection) of the combined join record. Note that setting the APPLICATION_SERVER_CONFIG parameter config doesnt actually tell Kafka Streams to start listening on whatever port you configure. For example, to implement a video game leaderboard, we need some way to compute the top three high scores for a given game. You can verify with the kafka-topics console script:13. This approach uses a local store to save state. However, this query type will return an iterator for all of the entries in our state store, instead of those within a specific key range only. Add a dedicated method for retrieving the state store that contains the leaderboard aggregations. A ValueJoiner simply takes each record that is involved in the join, and produces a new, combined record. // StreamsConfig.PROCESSING_GUARANTEE_CONFIG="processing.guarantee", // StreamsConfig.EXACTLY_ONCE="exactly_once", // ("alice", "I like coffee") -> ("alice", "I"), ("alice", "like"), ("alice", "coffee"), // KTable groupBy method returns KGroupedTable, // KStream groupBy method returns KGroupedStream, Continue reading How to enable remote access to MySQL, Pareando Dispositivo 2.4GHz com Modem Vivo Fibra Askey, Processing with External Lookup: Stream-Table Join, Choosing a Stream Processing Approach for Your Application, Kafka Streams Architecture: Simple and Flexible. If youre interested in the implementation details, please check the source code for this chapter. As with relational systems, Kafka Streams includes support for multiple kinds of joins. However, in addition to leveraging stateless operators to filter, branch, merge, and transform facts, we can ask even more advanced questions of our data if we learn how to model behaviors using stateful operators. To clarify the first point, a persistent state store may keep some of its state in-memory, while writing to disk when the size of the state gets too big (this is called spilling to disk) or when the write buffer exceeds a configured value. Some of the topics we will cover include: The benefits of stateful stream processing, The differences between facts and behaviors, What kinds of stateful operators are available in Kafka Streams, How state is captured and queried in Kafka Streams, How the KTable abstraction can be used to represent local, partitioned state, How the GlobalKTable abstraction can be used to represent global, replicated state, How to perform stateful operations, including joining and aggregating data, How to use interactive queries to expose state. The input topics on both sides of the join must contain the same number of partitions. Running a single Kafka Streams application would consolidate the entire application state to a single instance, but Kafka Streams is meant to be run in a distributed fashion for maximizing performance and fault tolerance. In this particular case, where were handling a group by aggregation, we can assume that all related messages are in the same partition of a same topic (because of the same key). Request-response patterns are better to suit this particular need. The process of aggregating tables is pretty similar to aggregating streams. Lets consider the following flow of events: If the step 3 do not occur (messages are being processed but not confirmed to the clients), the retry mechanism will make clients to send the messages again. Lets take a look at the abstraction that allows us to do this: GlobalKTable. With regards to the keyspace, players contains a record for each unique player in our system.