---
metadata:
  - name: generator
    content: Diplodoc Platform v5.56.0
alternate:
  - https://ydb.tech/docs/en/dev/streaming-query/patterns.md?version=v26.2
  - https://ydb.tech/docs/ru/dev/streaming-query/patterns.md?version=v26.2
  - href: en/dev/streaming-query/patterns.md
    type: text/markdown
    title: Markdown version
  - href: ../../llms.txt
    type: text/markdown
    title: llms.txt
sourcePath: en/core/dev/streaming-query/patterns.md
---
> **Documentation Index:** Fetch the complete configuration index at https://ydb.tech/docs/en/llms.txt

# Typical streaming query patterns

This section contains minimal examples of streaming queries for the most common scenarios. First, a basic pattern for reading data from a topic is described, followed by variants of full data processing: data processing and writing results to a topic in JSON format, to a topic as a string, and to a table. Each example can be used as a starting point for your own tasks.

The examples below use [local and external topics](https://ydb.tech/docs/en/concepts/query_execution/topics.md?version=v26.2#local-external-topics). Notation:

- `ext_source` — a pre-created [`external data source`](https://ydb.tech/docs/en/concepts/datamodel/external_data_source.md?version=v26.2)
- `input_topic` — the topic from which data is read
- `output_topic` — the topic where results are written
- `output_table` — the YDB table where results are written

## Reading data from a topic {#topic-read}

Reading structured messages is done using `SELECT ... FROM ... WITH (FORMAT, SCHEMA)`. The `WITH` block specifies the input data format and schema — which fields are expected in each message and their types.

The following fragment is used inside [CREATE STREAMING QUERY](https://ydb.tech/docs/en/yql/reference/syntax/create-streaming-query.md?version=v26.2) in the `DO BEGIN ... END DO` block:


```yql
SELECT
    Id,
    Name
FROM
    topic_name  -- local topic; for external: ext_source.topic_name
WITH (
    FORMAT = json_each_row,
    SCHEMA = (
        Id Uint64 NOT NULL,
        Name Utf8 NOT NULL
    )
);
```


For more details on formats: [Data formats for reading/writing from topics](https://ydb.tech/docs/en/dev/streaming-query/streaming-query-formats.md?version=v26.2).

This pattern is used in all subsequent examples.

## Writing to a topic (JSON) {#topic-json}

The query reads events from the input topic, forms a JSON object from individual fields, and writes the result to the 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 the topic.


```yql
CREATE STREAMING QUERY write_json_example AS
DO BEGIN

INSERT INTO ext_source.output_topic -- or local topic output_topic
SELECT
    -- Building JSON from individual fields
    ToBytes(Unwrap(Yson::SerializeJson(Yson::From(
        AsStruct(Id AS id, Name AS 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
```


For more details on functions:

- [AsStruct](https://ydb.tech/docs/en/yql/reference/builtins/basic.md?version=v26.2#as-container)
- [Yson::From](https://ydb.tech/docs/en/yql/reference/udf/list/yson.md?version=v26.2#ysonfrom)
- [Yson::SerializeJson](https://ydb.tech/docs/en/yql/reference/udf/list/yson.md?version=v26.2#ysonserializejson)
- [Unwrap](https://ydb.tech/docs/en/yql/reference/builtins/basic.md?version=v26.2#unwrap)
- [ToBytes](https://ydb.tech/docs/en/yql/reference/builtins/basic.md?version=v26.2#to-from-bytes).

## Writing to a topic (string) {#topic-utf8}

The query reads events from the input topic and writes one field as a string to the output topic. To write strings to a topic, `SELECT` must return a single column of type `String` or `Utf8`.


```yql
CREATE STREAMING QUERY write_utf8_example AS
DO BEGIN

INSERT INTO output_topic -- or external topic ext_source.output_topic
SELECT
    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
```


For more details on write formats: [Formats for writing data](https://ydb.tech/docs/en/dev/streaming-query/streaming-query-formats.md?version=v26.2#write_formats).

## Writing to a table {#table-write}

The query reads events from the 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 (the at-least-once guarantee) it would lead to duplicate rows. With `UPSERT`, if a row with such a primary key already exists, it will be updated; otherwise, a new row will be inserted, and `INSERT INTO` will fail with an error.

{% endnote %}


```yql
CREATE STREAMING QUERY write_table_example AS
DO BEGIN

-- Writing to table (only UPSERT, INSERT not supported)
UPSERT INTO output_table
SELECT
    Id,
    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
```


For more details: [Writing to tables](https://ydb.tech/docs/en/dev/streaming-query/table-writing.md?version=v26.2).

## See also

- [Local and external topics](https://ydb.tech/docs/en/concepts/query_execution/topics.md?version=v26.2#local-external-topics)
- [CREATE STREAMING QUERY](https://ydb.tech/docs/en/yql/reference/syntax/create-streaming-query.md?version=v26.2)
- [Quickstart: reading and writing topics](https://ydb.tech/docs/en/recipes/streaming_queries/topics.md?version=v26.2)