---
sidebar_position: 3
---

# Persistent queries

Besides STREAM and TABLE definitions, in ksqlDB there are three kinds of queries:

- **Persistent**: These are server-side queries that run indefinitely processing rows of events.
- **Push**: These are client-side queries that subscribe to a result as it changes in real-time over a long-lived connection.
- **Pull**: These are client-side queries that retrieve a result as of "now", like a normal query against a traditional RDBMS.

Currently we only support the creation of persistent queries in ksqlDB Server.
There are two types of persistent queries:

- **CSAS**, or `CREATE STREAM AS SELECT`: defines a materialized stream view by constantly running a query and producing the results to a sink Kafka topic.
- **CTAS**, or `CREATE TABLE AS SELECT`: defines a materialized table view by constantly updating or aggregating records by their key into a sink Kafka topic.

You can review the detailed developer reference for both [CSAS](https://docs.ksqldb.io/en/latest/developer-guide/ksqldb-reference/create-stream-as-select/) and [CTAS](https://docs.ksqldb.io/en/latest/developer-guide/ksqldb-reference/create-table-as-select/) in the ksqlDB official documentation.

## Writing persistent queries

### Pay attention to the KEYS

When creating persistent queries, always try to ensure that you define KEY columns, or partition keys, as this helps ksqlDB optimize the query execution by leveraging the underlying Kafka topic partitions.

Keys are also very important in queries that involve **JOINs** or **AGGREGATIONs**, whenever possible you should define keys that benefit your queries, including the source topics/streams/tables. This ensures that related records with the same key are processed together, in an optimal way, and avoids forcing ksqlDB to perform an extra internal repartitioning to fit the query plan.

- Key streams and tables to fit your workload, using `KEY` columns or `GROUP BY`, `PARTITION BY` clauses.
- Key aggregation inputs on the grouping criteria. If you are aggregating using a single grouping column, consider keying the input stream on that column to avoid internal repartitioning.

### State Management and large time windows

Persistent queries maintain state internally for processing ongoing data. Keep this in mind as it's important to understand for the potential impact of state size and scalability, especially for long-running queries, and queries aggregating over large windows of time.

Dropping and recreating a stream or table might also incur in data loss on streams configured to aggregate over a long period of time, as the retained history of messages that exists within internal Kafka topics could get erased.


## Examples

### Creating a Stream from a Kafka topic
CSAS and CTAS queries require input streams or tables, so always define your stream(s) or table(s) from a topic first:

```sql
CREATE STREAM user_events_stream (
    `userid` VARCHAR,
    `event_type` VARCHAR,
    `timestamp` BIGINT
) WITH (KAFKA_TOPIC='user_events_topic', VALUE_FORMAT='JSON');
```

### Creating a Table from a Stream with Aggregations
```sql
CREATE TABLE user_click_counts AS
SELECT
    `userid`,
    COUNT(*) AS `click_count`
FROM user_events_stream
WHERE event_type = 'click'
GROUP BY `userid`;
```

### Join Streams to enrich data
```sql
CREATE STREAM enriched_events AS
SELECT
    e.userid,
    e.event_type,
    u.username
FROM user_events_stream e
LEFT JOIN user_profiles_table u
ON e.userid = u.userid;
```

### Window aggregations

Here are just a few examples of windowed aggregations that can be performed in ksqlDB.
For a more detailed explanation visit: [Time and Windows in ksqlDB queries](https://docs.ksqldb.io/en/latest/concepts/time-and-windows-in-ksqldb-queries/#windows-in-sql-queries).

1. **Tumbling windows**:
    This type of windowed aggregation partitions the stream into **fixed-size, non-overlapping, gap-less time intervals**, or "windows".
    Each window represents a discrete time period, and aggregates are computed independently for each window.

    ```sql
    CREATE TABLE click_counts_by_hour AS
    SELECT
        WINDOWSTART() AS window_start,
        COUNT(*) AS click_count
    FROM user_events_stream
    WINDOW TUMBLING (SIZE 1 HOUR)
    WHERE event_type = 'click'
    GROUP BY TUMBLINGWINDOW(1 HOUR);
    ```

2. **Hopping windows**:
    This type of windowed aggregation partitions the stream into **fixed-size, overlapping time intervals**.
    They are defined by their window's duration and its advance, or "hop", interval. Here a record can belong to more than one window.

    ```sql
    CREATE TABLE avg_clicks_over_time AS
    SELECT
        userid,
        SUM(purchase_total) AS total_revenue,
        WINDOWSTART() AS window_start,
        WINDOWEND() AS window_end,
    FROM user_events_stream
    WINDOW HOPPING (SIZE 30 SECONDS, ADVANCE BY 10 SECONDS)
    WHERE event_type = 'purchase'
    GROUP BY suserid;
    ```

3. **Session windows**:
    This type of windowed aggregation puts records together into a "session": a period of activity separated by a specified gap of inactivity, or "idleness".

    ```sql
    CREATE TABLE user_sessions AS
    SELECT
        userid,
        WINDOWSTART() AS session_start,
        WINDOWEND() AS session_end,
        COUNT(*) AS click_count
    FROM user_events_stream
    WINDOW SESSION (30 MINUTES)
    GROUP BY userid;
    ```