---
metadata:
  - name: generator
    content: Diplodoc Platform v5.50.6
alternate:
  - https://ydb.tech/docs/en/yql/reference/syntax/create-streaming-query.md?version=main
  - https://ydb.tech/docs/ru/yql/reference/syntax/create-streaming-query.md?version=main
sourcePath: en/core/yql/reference/syntax/create-streaming-query.md
---
> **Documentation Index:** Fetch the complete configuration index at https://ydb.tech/docs/en/llms.txt

# CREATE STREAMING QUERY

`CREATE STREAMING QUERY` creates a [streaming query](https://ydb.tech/docs/en/concepts/streaming-query/streaming-query.md?version=main).

## Syntax


```sql
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](https://ydb.tech/docs/en/concepts/streaming-query/streaming-query.md?version=main#limitations), see [below](#examples) 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](https://ydb.tech/docs/en/concepts/glossary.md?version=main#resource-pool) in which the query will run.

See [below](#examples) for examples of creating a streaming query.

## Using a Consumer {#consumer-usage}

A [consumer](https://ydb.tech/docs/en/concepts/datamodel/topic.md?version=main#consumer) is a named subscription to a [topic](https://ydb.tech/docs/en/concepts/datamodel/topic.md?version=main) that stores the current read position.

A consumer is created via the [CLI](https://ydb.tech/docs/en/reference/ydb-cli/topic-consumer-add.md?version=main) or when creating a topic using [CREATE TOPIC](https://ydb.tech/docs/en/yql/reference/syntax/create-topic.md?version=main). The consumer name is specified in the query text using a pragma:


```sql
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](https://ydb.tech/docs/en/dev/streaming-query/checkpoints.md?version=main). Specifying a consumer allows tracking the read position and lag from the topic side, for example, via the [CLI](https://ydb.tech/docs/en/reference/ydb-cli/topic-read.md?version=main).

## Examples {#examples}

### Writing to a Topic (JSON) {#example-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 info %}

Streaming queries can work with [local and external topics](https://ydb.tech/docs/en/dev/streaming-query/local-and-external-topics.md?version=main).

In the example:

- `ext_source` is a pre-created [`external data source`](https://ydb.tech/docs/en/concepts/datamodel/external_data_source.md?version=main).
- `input_topic` and `output_topic` are local or external [topics](https://ydb.tech/docs/en/concepts/datamodel/topic.md?version=main).

{% endnote %}


```yql
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 {#example-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.

{% note 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](https://ydb.tech/docs/en/concepts/streaming-query/streaming-query.md?version=main#guarantees)), 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.

{% endnote %}


```sql
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 {#example-resource-pool}

The query is created in the specified [resource pool](https://ydb.tech/docs/en/concepts/glossary.md?version=main#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](https://ydb.tech/docs/en/yql/reference/syntax/alter-streaming-query.md?version=main).


```sql
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](https://ydb.tech/docs/en/dev/streaming-query/patterns.md?version=main).

## See also

* [Common streaming query patterns](https://ydb.tech/docs/en/dev/streaming-query/patterns.md?version=main)
* [Streaming queries](https://ydb.tech/docs/en/concepts/streaming-query/streaming-query.md?version=main)
* [ALTER STREAMING QUERY](https://ydb.tech/docs/en/yql/reference/syntax/alter-streaming-query.md?version=main)
* [DROP STREAMING QUERY](https://ydb.tech/docs/en/yql/reference/syntax/drop-streaming-query.md?version=main)