ksqlDB Concepts

🧠 Free Recall ❓ Quiz 🔍 Cues 🔮 Simulation 🥷 ShadowBox 📐 Encode Cards

What is an Event

ksqlDB is a database that's purpose-built for stream processing applications. The main focus of stream processing is modeling computation over unbounded streams of events.

An event is anything that occurred and was recorded. It could be something high-level that happened in a business, like the sale of an item or the submission of an invoice. Or it could be something low-level, like a log line emitted by a web server when a request is received. Anything that happens at a point in time is an event.

Because events are so fundamental to stream processing, they are ksqlDB's core unit of data. All of ksqlDB's features are oriented around making it easy to solve problems using events. Although it's easy to think about individual events, figuring out how to store related events together is a bit more challenging. Fortunately, the idea of storing related events is well-explored territory. Apache Kafka® leads the way, which is why ksqlDB is built directly on top of it.

Kafka is a distributed streaming platform for working with events. It’s horizontally scalable, fault-tolerant, and extremely fast. Although working with it directly can be low-level, it has a strong and opinionated approach for modeling both individual events and stored events. For this reason, ksqlDB borrows heavily from some of Kafka's abstractions. It doesn’t aim to make you learn all of Kafka, but it also doesn't reinvent the wheel where there's already something really good to use.

ksqlDB represents events by using a simple key/value model, which is very similar to Kafka's notion of a record. The key represents some form of identity about the event. The value represents information about the event that occurred. This combination of key and value makes it easy to model stored events, since multiple events with the same key represent the same identity, irrespective of their values.

But events in ksqlDB carry more information than just a key and value. Similar to Kafka, they also describe the time at which the event was true.

ksqlDB aims to raise the abstraction from working with a lower-level stream processor. Usually, an event is called a "row", as if it were a row in a relational database. Each row is composed of a series of columns. Columns are either read from the event's key or value. ksqlDB also supports 3 pseudo columns, available on every row: ROWTIME, which represents the time of the event, as well as ROWPARTITION and ROWOFFSET, which represent the partition and offset of the source record, respectively. In addition, windowed sources have WINDOWSTART and WINDOWEND system columns.

Starting in ksqlDB 0.24, you can create columns that are populated by a record's header field. Kafka headers are a list of zero or more key-value pairs, where the keys are (not necessarily unique) strings and the values are byte arrays. Headers typically contain metadata about records, which can then be used for routing or processing For example, the header could store information about how to deserialize strings in the value.

Stream Processing

Creating a collection with an enforced schema over a new or existing Apache Kafka® topic is useful but has limited utility by itself for creating an application. When you declare a collection, you can only work with the events in their current form. But a critical part of creating streaming applications is transforming, filtering, joining, and aggregating events.

In ksqlDB, you manipulate events by deriving new collections from existing ones and describing the changes between them. When a collection is updated with a new event, ksqlDB updates the collections that are derived from it in real-time. This rich form of computing is known formally as stream processing, because it creates programs that operate continually over unbounded streams of events, ad infinitum. These processes stop only when you explicitly terminate them.

The general pattern for stream processing in ksqlDB is to create a new collection by using the SELECT statement on an existing collection. The result of the inner SELECT feeds into the outer declared collection. You don't need to declare a schema when deriving a new collection, because ksqlDB infers the column names and types from the inner SELECT statement. The value of the ROWTIME pseudo column defines the timestamp of the record written to Kafka, and the value of the ROWPARTITION and ROWOFFSET pseudo columns define the partition and offset of the source record, respectively. The value of system columns can not be set in the SELECT.

Materialized Views

Materialized Views
In any database, one of the main purposes of a table is to enable efficient queries over the data. ksqlDB stores events immutably in Apache Kafka® by using a simple key/value model. But how can queries be made efficient under this model? The answer is by leveraging materialized views.

Stream/Table Duality
Streams and tables are closely related. A stream is a sequence of events that you can derive a table from. For example, a sequence of credit scores for a loan applicant can change over time. The sequence of credit scores is a stream. But this stream can be interpreted as a table to describe the applicant's current credit score.
Conversely, the table that represents current credit scores is really two things: the current credit scores, and also the sequence of changes to the credit scores for each applicant. This is a profound realization, and much has been written on this stream/table duality. For more information, see Streams and Tables: Two Sides of the Same Coin.
Traditional databases have redo logs, but subscribing to changes can be cumbersome. Redo logs have much shorter retention than the Kafka changelog topic. A fully compacted Kafka changelog topic is the same as a database snapshot. Efficient queries evaluate just the changes.

Materialized Views
The benefit of a materialized view is that it evaluates a query on the changes only (the delta), instead of evaluating the query on the entire table.
When a new event is integrated, the current state of the view evolves into a new state. This transition happens by applying the aggregation function that defines the view with the current state and the new event. When a new event is integrated, the aggregation function that defines the view is applied only on this new event, leading to a new state for the view. In this way, a view is never "fully recomputed" when new events arrive. Instead, the view adjusts incrementally to account for the new information, which means that queries against materialized views are highly efficient.
In ksqlDB, a table can be materialized into a view or not. If a table is created directly on top of a Kafka topic, it's not materialized. Non-materialized tables can't be queried, because they would be highly inefficient. On the other hand, if a table is derived from another collection, ksqlDB materializes its results, and you can make queries against it.
ksqlDB leverages the idea of stream/table duality by storing both components of each table. The current state of a table is stored locally and ephemerally on a specific server by using RocksDB. The series of changes that are applied to a table is stored durably in a Kafka topic and is replicated across Kafka brokers. If a ksqlDB server with a materialization of a table fails, a new server rematerializes the table from the Kafka changelog.

Streams

A stream is a partitioned, immutable, append-only collection that represents a series of historical facts. For example, the rows of a stream could model a sequence of financial transactions, like "Alice sent $100 to Bob", followed by "Charlie sent $50 to Bob".
Once a row is inserted into a stream, it can never change. New rows can be appended at the end of the stream, but existing rows can never be updated or deleted.
Each row is stored in a particular partition. Every row, implicitly or explicitly, has a key that represents its identity. All rows with the same key reside in the same partition.

Tables

A table is a mutable, partitioned collection that models change over time. In contrast with a stream, which represents a historical sequence of events, a table represents what is true as of "now". For example, you might use a table to model the locations where someone has lived as a stream: first Miami, then New York, then London, and so forth.
Tables work by leveraging the keys of each row. If a sequence of rows shares a key, the last row for a given key represents the most up-to-date information for that key's identity. A background process periodically runs and deletes all but the newest rows for each key.

Queries

There are three kinds of queries in ksqlDB: persistent, push, and pull. Each gives you different ways to work with rows of events.

Persistent¶
Persistent queries are server-side queries that run indefinitely processing rows of events. You issue persistent queries by deriving new streams and new tables from existing streams or tables.

Push¶

A push query is a form of query issued by a client that subscribes to a result as it changes in real-time. A good example of a push query is subscribing to a particular user's geographic location. The query requests the map coordinates, and because it's a push query, any change to the location is "pushed" over a long-lived connection to the client as soon as it occurs. This is useful for building programmatically controlled microservices, real-time apps, or any sort of asynchronous control flow.
Push queries are expressed using a SQL-like language. They can be used to query either streams or tables for a particular key. Also, push queries aren't limited to key look-ups. They support a full set of SQL, including filters, selects, group bys, partition bys, and joins.
Push queries enable you to query a stream or materialized table with a subscription to the results. You can subscribe to the output of any query, including one that returns a stream. A push query emits refinements to a stream or materialized table, which enables reacting to new information in real-time. They’re a good fit for asynchronous application flows. For request/response flows, see pull Query.
Execute a push query by sending an HTTP request to the ksqlDB REST API, and the API sends back a chunked response of indefinite length.
The result of a push query isn't persisted to a backing Kafka topic. If you need to persist the result of a query to a Kafka topic, use a CREATE TABLE AS SELECT or CREATE STREAM AS SELECT statement.

Pull¶

A pull query is a form of query issued by a client that retrieves a result as of "now", like a query against a traditional RDBMS.
As a dual to the push query example, a pull query for a geographic location would ask for the current map coordinates of a particular user. Because it's a pull query, it returns immediately with a finite result and closes its connection. This is ideal for rendering a user interface once, at page load time. It's generally a good fit for any sort of synchronous control flow.
Pull queries enable you to fetch the current state of a materialized view. Because materialized views are incrementally updated as new events arrive, pull queries run with predictably low latency. They're a great match for request/response flows. For asynchronous application flows, see Push Query. Pull queries are expressed using a strict subset of ANSI SQL.
Execute a pull query by sending an HTTP request to the ksqlDB REST API, and the API responds with a single response.
If you want to learn more about Pull Queries, see Pull Queries.

Joins

You can use ksqlDB to merge streams of events in real time by using the JOIN statement, which has a SQL join syntax. A ksqlDB join and a relational database join are similar in that they both combine data from two or more sources based on common values. The result of a ksqlDB join is a new stream or table that's populated with the column values that you specify in a SELECT statement.
With ksqlDB, you don't need to write the low-level logic around joining streams and tables, so you can focus on the business logic for combining your streaming data.
You can join streams and tables in these ways:

Join multiple streams to create a new stream.
Join multiple tables to create a new table.
Join multiple streams and tables to create a new stream.

ksqlDB enables grouping records that have the same key for stateful operations, like joins, into windows. At the moment, ksqlDB uses a fixed grace period of 24 hours, which means that a record can arrive out-of-order for up to 24 hours, and it's still joined correctly based on its timestamp. If a record arrives after the window's grace period has passed, the record is discarded and isn't processed.

Note
Only stream-stream joins are windowed.
Windows are tracked per record key. In join operations, ksqlDB uses a windowing state store to store all of the records received so far within the defined window boundary. Old records in the state store are purged after the window grace period passed.

Time and Windows

In ksqlDB, a record is an immutable representation of an event in time. Each record carries a timestamp, which determines its position on the time axis.
This is the default timestamp that ksqlDB uses for processing the record. The timestamp is set either by the producer application or by the Apache Kafka® broker, depending on the topic's configuration. Records may be out-of-order within the stream.
Timestamps are used by time-dependent operations, like aggregations and joins.

Time semantics¶
Timestamps have different meanings, depending on the implementation. A record's timestamp can refer to the time when the event occurred, or when the record was ingested into Kafka, or when the record was processed. These times are event-time, ingestion-time, and processing-time.

Event-time¶
The time when a record is created by the data source. Achieving event-time semantics requires embedding timestamps in records when an event occurs and the record is produced.
For example, if the record is a geo-location change reported by a GPS sensor in a car, the associated event-time is the time when the GPS sensor captured the location change.

Ingestion-time¶
The time when a record is stored in a topic partition by a Kafka broker. Ingestion-time is similar to event-time, as a timestamp is embedded in the record, but the ingestion timestamp is generated when the Kafka broker appends the record to the target topic.
Ingestion-time can approximate event-time if the time difference between the creation of the record and its ingestion into Kafka is small.
For use cases where event-time semantics aren't possible, ingestion-time may be an alternative. Consider using ingestion-time when data producers don't embed timestamps in records, as in older versions of Kafka's Java producer client, or when the producer can't assign timestamps directly, like when it doesn't have access to a local clock.

Processing-time¶
The time when the record is consumed by a stream processing application. The processing-time can occur immediately after ingestion-time, or it may be delayed by milliseconds, hours, days, or longer.
For example, imagine an analytics application that reads and processes the geo-location data reported from car sensors, and presents it to a fleet-management dashboard. In this case, processing-time in the analytics application might be many minutes or hours after the event-time, as cars can move out of mobile reception for periods of time and have to buffer records locally.

Stream-time¶
The maximum timestamp seen over all processed records so far.

Important
Don't mix streams or tables that have different time semantics.
Timestamp assignment¶
A record's timestamp is set either by the record's producer or by the Kafka broker, depending on the topic's timestamp configuration. The topic's message.timestamp.type setting can be either CreateTime or LogAppendTime.

CreateTime: The broker uses the record's timestamp as set by the producer. This setting enforces event-time semantics.
LogAppendTime: The broker overwrites the record's timestamp with the broker's local time when it appends the record to the topic's log. This setting enforces ingestion-time semantics. If LogAppendTime is configured, the producer has no control over the timestamp.
ksqlDB doesn't support processing-time operations directly, but you can implement user-defined functions (UDFs) that access the current time. For more information, see Functions.
By default, when ksqlDB imports a topic to create a stream, it uses the record's timestamp, but you can add the WITH(TIMESTAMP='some-field') clause to use a different field from the record's value as the timestamp. The optional TIMESTAMP_FORMAT property indicates how ksqlDB should parse the field. The field you specify can be an event-time or an ingestion-time. This approach implements payload-time semantics.

Important
If you use the WITH(TIMESTAMP=...) clause, this timestamp must be expressible as a Unix epoch time in milliseconds, which is the number of milliseconds that have elapsed since 1 January 1970 at midnight UTC/GMT. Also, you can specify the timestamp as a string when you provide a TIMESTAMP_FORMAT. For more information, see Timestamp Formats.
When working with time you should also make sure that additional aspects of time, like time zones and calendars, are correctly synchronized — or at least understood and traced — throughout your streaming data pipelines. It helps to agree on specifying time information in UTC or in Unix time, like seconds since the Unix epoch, everywhere in your system.

Timestamps of ksqlDB output streams¶
When a ksqlDB application writes new records to Kafka, timestamps are assigned to the records it creates. ksqlDB uses the underlying Kafka Streams implementation for computing timestamps. Timestamps are assigned based on context:

When new output records are generated by processing an input record directly, output record timestamps are inherited from input record timestamps.
When new output records are generated by a periodic function, the output record timestamp is defined as the current internal time of the stream task.
For stateless operations, the input record timestamp is passed through. For flatMap and siblings that emit multiple records, all output records inherit the timestamp from the corresponding input record.
Timestamps for aggregations and joins¶
For aggregations and joins, timestamps are computed by using the following rules.

For joins (stream-stream, table-table) that have left and right input records, the timestamp of the output record is assigned max(left.ts, right.ts).
For stream-table joins, the output record is assigned the timestamp from the stream record.
For aggregations, the max timestamp is computed over all records, per key, either globally (for non-windowed) or per-window.
Producers and timestamps¶
A producer application can set the timestamp on its records to any value, but usually, it choses a sensible event-time or the current wall-clock time.
If the topic's message.timestamp.type configuration is set to CreateTime, the following holds for the producer:

When a producer record is created, it contains no timestamp, by default.
The producer can set the timestamp on the record explicitly.
If the timestamp isn't set when the producer application calls the producer.send() method, the current wall-clock time is set automatically.
In all three cases, the time semantics are considered to be event-time.

Timestamp extractors¶
When ksqlDB imports a topic to create a stream, it gets the timestamp from the topic's messages by using a timestamp extractor class. Timestamp extractors implement the TimestampExtractor interface.
Concrete implementations of timestamp extractors may retrieve or compute timestamps based on the actual contents of data records, like an embedded timestamp field, to provide event-time or ingestion-time semantics, or they may use any other approach, like returning the current wall-clock time at the time of processing to implement processing-time semantics.
By creating a custom timestamp extractor class, you can enforce different notions or semantics of time, depending on the requirements of your business logic. For more information see default.timestamp.extractor.

Windows in SQL queries¶
Representing time consistently enables aggregation operations on streams and tables, like SUM, that have distinct time boundaries. In ksqlDB, these boundaries are named windows.

A window has a start time and an end time, which you access in your queries by using the WINDOWSTART and WINDOWEND system columns.

Important
ksqlDB is based on the Unix epoch time in the UTC timezone, and this can affect time windows. For example, if you define a 24-hour tumbling time window, it will be in the UTC timezone, which may not be appropriate if you want to have daily windows in your timezone.
Windowing lets you control how to group records that have the same key for stateful operations, like aggregations or joins, into time spans. ksqlDB tracks windows per record key.

Note
A related operation is grouping, which groups all records that have the same key to ensure that records are properly partitioned, or "keyed", for subsequent operations. When you use the GROUP BY clause in a query, windowing enables you to further sub-group the records of a key.
Windowing queries must group by the keys that are selected in the query.
When using windows in your SQL queries, aggregate functions are applied only to the records that occur within a specific time window. Records that arrive out-of-order are handled as you might expect: although the window end time has passed, the out-of-order records are still associated with the correct window.

Window types¶
There are three ways to define time windows in ksqlDB: hopping windows, tumbling windows, and session windows. Hopping and tumbling windows are time windows, because they are defined by fixed durations that you specify. Session windows are dynamically sized based on incoming data and defined by periods of activity separated by gaps of inactivity.

Window typeBehaviorDescriptionTumbling WindowTime-basedFixed-duration, non-overlapping, gap-less windowsHopping WindowTime-basedFixed-duration, overlapping windowsSession WindowSession-basedDynamically-sized, non-overlapping, data-driven windows

Hopping window¶
Hopping windows are based on time intervals. They model fixed-sized, possibly overlapping windows. A hopping window is defined by two properties: the window's duration and its advance, or "hop", interval. The advance interval specifies how far a window moves forward in time relative to the previous window. For example, you can configure a hopping window with a duration of five minutes and an advance interval of one minute. Because hopping windows can overlap, and usually they do, a record can belong to more than one such window.
All hopping windows have the same duration, but they might overlap, depending on the length of time specified in the ADVANCE BY property.

For example, if you want to count the pageviews for only Region_6 by female users for a hopping window of 30 seconds that advances by 10 seconds, you might run a query like this:

1
2
3
4
5
SELECT regionid, COUNT(*) FROM pageviews
  WINDOW HOPPING (SIZE 30 SECONDS, ADVANCE BY 10 SECONDS)
  WHERE UCASE(gender)='FEMALE' AND LCASE (regionid) LIKE '%_6'
  GROUP BY regionid
  EMIT CHANGES;
The hopping window's start time is inclusive, but the end time is exclusive. This is important for non-overlapping windows, in which each record must be contained in exactly one window.

Tumbling window¶
Tumbling windows are a special case of hopping windows. Like hopping windows, tumbling windows are based on time intervals. They model fixed-size, non-overlapping, gap-less windows. A tumbling window is defined by a single property: the window's duration. A tumbling window is a hopping window whose window duration is equal to its advance interval. Since tumbling windows never overlap, a record will belong to one and only one window.
All tumbling windows are the same size and adjacent to each other, which means that whenever a window ends, the next window starts.

For example, if you want to compute the five highest-value orders per zip code per hour in an orders stream, you might run a query like this:

1
2
3
SELECT orderzip_code, TOPK(order_total, 5) FROM orders
  WINDOW TUMBLING (SIZE 1 HOUR) GROUP BY order_zipcode
  EMIT CHANGES;
Here's another example: to detect potential credit card fraud in an authorization_attempts stream, you might run a query for the number of authorization attempts on a particular card that's greater than three, during a time interval of five seconds.

1
2
3
4
SELECT card_number, count(*) FROM authorization_attempts
  WINDOW TUMBLING (SIZE 5 SECONDS)
  GROUP BY card_number HAVING COUNT(*) > 3
  EMIT CHANGES;
The tumbling window's start time is inclusive, but the end time is exclusive. This is important for non-overlapping windows, in which each record must be contained in exactly one window.

Session window¶
A session window aggregates records into a session, which represents a period of activity separated by a specified gap of inactivity, or "idleness". Any records with timestamps that occur within the inactivity gap of existing sessions are merged into the existing sessions. If a record's timestamp occurs outside of the session gap, a new session is created.
A new session window starts if the last record that arrived is further back in time than the specified inactivity gap.
Session windows are different from the other window types, because:

ksqlDB tracks all session windows independently across keys, so windows of different keys typically have different start and end times.
Session window durations vary. Even windows for the same key typically have different durations.
Session windows are especially useful for user behavior analysis. Session-based analyses range from simple metrics, like counting user visits on a news website or social platform, to more complex metrics, like customer-conversion funnel and event flows.

For example, to count the number of pageviews per region for session windows with a session inactivity gap of 60 seconds, you might run the following query, which sessionizes the input data and performs the counting/aggregation step per region:

1
2
3
4
SELECT regionid, COUNT(*) FROM pageviews
  WINDOW SESSION (60 SECONDS)
  GROUP BY regionid
  EMIT CHANGES;
The start and end times for a session window are both inclusive, in contrast to time windows.
A session window contains at least one record. It's not possible for a session window to have zero records.
If a session window contains exactly one record, the record's ROWTIME timestamp is identical to the window's own start and end times. Access these by using the WINDOWSTART and WINDOWEND system columns.
If a session window contains two or more records, then the earliest/oldest record's ROWTIME timestamp is identical to the window's start time, and the latest/newest record's ROWTIME timestamp is identical to the window's end time.

Windowed joins¶
ksqlDB supports using windows in JOIN queries by using the WITHIN clause.
For example, to find orders that have shipped within the last hour from an orders stream and a shipments stream, you might run a query like:

1
2
3
4
5
6
SELECT o.order_id, o.total_amount, o.customer_name, s.shipment_id, s.warehouse
  FROM new_orders o
  INNER JOIN shipments s
    WITHIN 1 HOURS
    ON o.order_id = s.order_id
    EMIT CHANGES;
For more information on joins, see Join Event Streams with ksqlDB.

Out-of-order events¶
Frequently, events that belong to a window can arrive out-of-order, for example, over slow networks, and a grace period may be required to ensure the events are accepted into the window. ksqlDB enables configuring this behavior for each of the window types and uses a default of 24 hours when no grace period is specified. Set this value explicitly to fit your scenario.
For example, to allow events to be accepted for up to two hours after the window ends, you might run a query like:

1
2
3
4
SELECT orderzip_code, TOPK(order_total, 5) FROM orders
  WINDOW TUMBLING (SIZE 1 HOUR, GRACE PERIOD 2 HOURS) 
  GROUP BY order_zipcode
  EMIT CHANGES;
Events that arrive after the grace period has passed are called late and aren't included in the aggregation result.

User-defined functions

ksqlDB comes with a broad range of built-in functions out of the box. But what happens when you need to use a function that ksqlDB doesn't have?

User-defined functions let you add new functions using Java hooks. Learn how to use them in the how-to guide, or look up critical information about them in the reference documentation.

Connectors
Lambda functions
Apache Kafka Primer

Unstarred

Dunk (exclude from study)

ksqlDB Concepts

Review Questions

Todos

Sensemaker