--- sidebar_position: 2 --- # Streams and Tables Streams and Tables are the two main types of storage abstractions in ksqlDB. They are both backed by Kafka topics, from which ksqlDB infers the schemas. ## Streams A Stream in ksqlDB represents an append-only unbounded sequence of immutable records that represents a series of historical facts. Streams allow for continuous and real-time processing of data as it flows through Kafka topics. Every row in a Stream is stored with an implicit or explicit KEY that represents it's identity. All rows with the same key reside in the same partition. Streams allow for real-time processing of events in a topic, continuously producing messages to a new Stream topic with the results. ### When to use Streams - **Event Sourcing**: Capture and process events as they occur. - **Monitoring**: Monitor messages or logs for anomalies and errors in real-time. - **Aggregation**: Aggregate messages from various sources into a unified Stream. - **Message transformations**: Perform basic transformations over messages in a Kafka topic. ### Creating a Stream Creating a stream registers it on an underlying Kafka topic, so you can perform operations like joins and aggregations on the topic's data. There is two main ways to create a new Stream in ksqlDB: - **[CREATE STREAM statement](https://docs.ksqldb.io/en/latest/developer-guide/ksqldb-reference/create-stream/)**: Registers a new Stream on a specified Kafka topic. - **CSAS - [CREATE STREAM AS SELECT statement](https://docs.ksqldb.io/en/latest/developer-guide/ksqldb-reference/create-stream-as-select/)**: Registers a new materialized Stream view that gets populated using a ksqlDB query. The resulting stream is said to be a persistent query. #### Columns A stream can store data in 3 types of columns: - **KEY**: Columns defined as `KEY` are stored in the KEY columns. A stream's key doesn't have to be unique, and can hold `NULL` values. - **HEADER**: Columns defined as `HEADERS`/`HEADER('')` are stored as Header of the Kafka message. - **VALUE**: Columns that are not defined as KEY or HEADER are stored in the message value. ### Properties Use the `WITH` clause to specify details about your stream: - **FORMAT**: [Serialization format](https://docs.ksqldb.io/en/latest/reference/serialization/) for both message key and value in the topic. - **KAFKA_TOPIC**: The name of the Kafka topic that backs the stream. The topic must already exist in Kafka (recommended), or you must specify `PARTITIONS` to auto-create the topic. The statement fails if the topic exists already with different partition/replica counts. - **KEY_FORMAT**: Serialization format for the message key. Can't be used with `FORMAT`. - **KEY_SCHEMA_ID**: Schema ID of the Key Schema in Schema Registry. - **PARTITIONS**: The number of partitions in the backing topic. You must set this property if you create a stream without an existing topic. You can't change the number of partitions on an existing stream. - **REPLICAS**: The number of replicas in the backing topic. - **RETENTION_MS**: The retention specified in milliseconds in the backing topic. - **TIMESTAMP**: By default, the `ROWTIME` pseudo colum is the timestamp (or event time) of the message in the Kafka Topic. You can override this to specify a column from the message. - **TIMESTAMP_FORMAT**: Used with `TIMESTAMP`, to specify the type and format of the timestamp column. - **VALUE_DELIMITER**: Set the delimiter string to use when `VALUE_FORMAT` is set to `DELIMITED`. - **VALUE_FORMAT**: The serialization format of the message value in topic. Can't be used with `FORMAT`. - **VALUE_SCHEMA_ID**: The schema ID of the Value Schema in Schema Registry. - **WRAP_SINGLE_VALUE**: Specifies how ksqlDB deserializes the value of a message in the backing topic that contains only a single column. ## Tables A table in ksqlDB represents a continuously updating result set of a query. Unlike streams, tables are mutable, leveraging the **keys** of each row to keep only the most up-to-date value for each key in the topic. ### When to use Tables - **State tracking**: Capture the most up-to-date state of an entity or query. - **Aggregation Results**: Continuously compute and store aggregation results. - **Stateful alerts**: Generate alerts based on the state of a continuosuly updated table. - **Materialized views**: Create views of frequently queried data. ### Creating a Table Creating a table registers it on an underlying Kafka topic, so you can perform operations like joins and aggregations on the topic's data. There is two main ways to create a new Table in ksqlDB: - **[CREATE TABLE statement](https://docs.ksqldb.io/en/latest/developer-guide/ksqldb-reference/create-table/)**: Registers a new Table on a specified Kafka topic. - **CTAS - [CREATE TABLE AS SELECT statement](https://docs.ksqldb.io/en/latest/developer-guide/ksqldb-reference/create-table-as-select/)**: Registers a new materialized Table view that gets populated using a KSQL SELECT query. #### Primary KEY On a ksqlDB Table each row is identified by its `PRIMARY KEY`, and this can't be `NULL`. If an incoming message in the backing Kafka topic has the same key as an existing row, it replaces the existing row in the table, but if the message's value is `NULL`, it deletes the row (tombstone record). ### Columns A tabke can store data in 3 types of columns: - **PRIMARY KEY**: Columns defined as `PRIMARY KEY` are stored in the KEY columns. A table's key must be unique, can't hold `NULL` values. - **HEADER**: Columns defined as `HEADERS`/`HEADER('')` are stored as Header of the Kafka message. - **VALUE**: Columns that are not defined as KEY or HEADER are stored in the message value. ### Pseudocolumns Pseudocolumns are automatically populated by ksqlDB that contain meta-information that can be inferred about the row at creation time. These are not returned when selecting all columns, but can be explicitly selected. The four available pseudocolumns are: - `HEADERS`: Columns that are populated by the Kafka recrod's header. - `ROWOFFSET`: The offset of the source record. - `ROWPARTITION`: The partition of the source record. - `ROWTIME`: Row timestamp, inferred from the underlying Kafka record if not overriden. ## Developing new ksqlDB collections We maintain our instance of ksqlDB in [kafka-connect/ksqldb_server](https://github.com/theorchard/kafka-connect/tree/master/ksqldb_server), where all queries are defined per environment. Follow the instructions in the README.md file to start developing new queries.