CREATE STREAMING QUERY

CREATE STREAMING QUERY creates a streaming query.

Syntax

CREATE [OR REPLACE] STREAMING QUERY [IF NOT EXISTS] <query_name> [WITH (
    <key1> = <value1>,
    <key2> = <value2>,
    ...
)] AS
DO BEGIN
    <query_statement1>;
    <query_statement2>;
    ...
END DO

Parameters

OR REPLACE — if a streaming query with this name already exists, it will be replaced with a new query while preserving the read offsets from the topic.
IF NOT EXISTS — do not output an error if a streaming query with this name already exists; in this case, the existing query will remain unchanged.
query_name — the name of the streaming query to create.
WITH (<key> = <value>) — a list of settings for the new streaming query, optional.
AS DO BEGIN ... END DO — the full text of the new streaming query, including all necessary SQL statements. Restrictions for the query text are given in Limitations, see below for query text examples.

Settings OR REPLACE and IF NOT EXISTS cannot be used simultaneously.

Available parameters of the WITH block:

RUN = (TRUE|FALSE) — start the query after creation, default TRUE.
RESOURCE_POOL = <resource_pool_name> — the name of the resource pool in which the query will run.

See below for examples of creating a streaming query.

Using a Consumer

A consumer is a named subscription to a topic that stores the current read position.

A consumer is created via the CLI or when creating a topic using CREATE TOPIC. The consumer name is specified in the query text using a pragma:

PRAGMA pq.Consumer="my_consumer";

If no consumer is specified, reading from the topic is performed without a consumer. In both cases, the read position is saved in a checkpoint. Specifying a consumer allows tracking the read position and lag from the topic side, for example, via the CLI.

Examples

Writing to a Topic (JSON)

The query reads events from an input topic, forms a JSON object from individual fields, and writes the result to an output topic.

The AsStruct function creates a structure from the specified fields, Yson::From converts it to Yson, Yson::SerializeJson serializes it to a JSON string, and ToBytes converts it to the String type, which is required for writing to a topic.

Note

Streaming queries can work with local and external topics.

In the example:

ext_source is a pre-created external data source.
input_topic and output_topic are local or external topics.

CREATE STREAMING QUERY my_streaming_query AS
DO BEGIN

    INSERT INTO output_topic -- or external topic ext_source.output_topic
    SELECT
        -- Forming JSON from individual fields
        ToBytes(Unwrap(Yson::SerializeJson(Yson::From(
            AsStruct(Id AS id, Name AS name)
        ))))
    FROM
        ext_source.input_topic -- or local topic input_topic
    WITH (
        FORMAT = json_each_row,  -- Input data format
        SCHEMA = (               -- Input data schema
            Id Uint64 NOT NULL,
            Name Utf8 NOT NULL
        )
    );

END DO

Writing to a Table

The query reads events from a topic and writes them to the output_table table. The table must be created in advance with a schema matching the selected columns.

Warning

Writing to tables in streaming queries is supported only in UPSERT mode. The INSERT INTO operation is not supported because, during reprocessing of events (at-least-once guarantee at-least-once), it would lead to duplicate rows. With UPSERT, if a row with the same primary key already exists, it will be updated; otherwise, a new row will be inserted, and INSERT INTO will fail with an error.

CREATE STREAMING QUERY my_streaming_query AS
DO BEGIN

    -- Writing to table (only UPSERT, INSERT not supported)
    UPSERT INTO output_table
    SELECT
        Id,
        Name
    FROM
        input_topic -- or external topic ext_source.input_topic
    WITH (
        FORMAT = json_each_row,  -- Input data format
        SCHEMA = (               -- Input data schema
            Id Uint64 NOT NULL,
            Name Utf8 NOT NULL
        )
    );

END DO

Running in a Resource Pool

The query is created in the specified resource pool but is not started automatically (RUN = FALSE). This allows you to check the configuration before starting or start the query later via ALTER STREAMING QUERY.

CREATE STREAMING QUERY my_streaming_query WITH (
    RUN = FALSE,                      -- Do not start automatically
    RESOURCE_POOL = my_resource_pool  -- Resource pool for execution
) AS
DO BEGIN

    INSERT INTO output_topic -- or external topic ext_source.output_topic
    SELECT
        ToBytes(Unwrap(Yson::SerializeJson(Yson::From(
            AsStruct(Id AS id, Name AS name)
        ))))
    FROM
        ext_source.input_topic -- or local topic input_topic
    WITH (
        FORMAT = json_each_row,
        SCHEMA = (
            Id Uint64 NOT NULL,
            Name Utf8 NOT NULL
        )
    );

END DO

Other examples: Common streaming query patterns.