A ksqlDB stream is an unbounded sequence of independent events. A base stream is created directly over a backing Kafka topic, exposing that topic’s records to ksqlDB queries. Unlike a table, repeated keys do not update a prior row; each record remains a separate event. section 11.1 section 11.4 section 11.7
Define a base stream with CREATE STREAM, its columns, and a WITH clause that identifies the Kafka topic, partition count, and value format. The definition can also select an event-time field and its format instead of relying on the producer-record timestamp. section 11.2 section 11.4
Listing 11.22 Creating the user_activity stream
CREATE STREAM user_activity (first_name VARCHAR,
last_name VARCHAR,
activity VARCHAR,
event_time VARCHAR,
steps INT
) WITH (kafka_topic='user_activity',
partitions=4,
value_format='JSON',
timestamp = 'event_time',
timestamp_format = 'yyyy-MM-dd HH:mm:ss'
);
This definition maps JSON records from user_activity into named columns and uses event_time as the stream timestamp. section 11.4
When the backing topic has record keys:
- Declare a typed column with the KEY keyword to expose the Kafka record key. section 11.4
- Set KEY_FORMAT to describe the key encoding; KAFKA supports basic key types such as String, Long, and Integer. section 11.4
- Use keys to drive incoming-record partitioning; without keys, records are evenly distributed across partitions. section 11.4
Turn phrases into a continuous output stream
Create input_stream over src-topic, then create yelling AS SELECT UCASE(phrase) AS SHOUT FROM input_stream EMIT CHANGES. The persistent query continuously transforms arriving records and writes its results as a new stream. section 11.1 section 11.7
Base streams serve as source queries. They can feed persistent queries that write results to Kafka topics, push queries that stream ongoing results to clients, and aggregations that produce tables. Schema Registry formats can supply column names and types from an existing schema rather than requiring explicit declarations. section 11.2 section 11.7
