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

# Common streaming query patterns

This section collects minimal examples of [streaming queries](https://ydb.tech/docs/en/concepts/streaming-query/streaming-query.md?version=main) for typical scenarios. It starts with a basic topic read, then shows end-to-end processing: handling data and writing results to a topic as JSON, to a topic as a plain string, and to a table. Each example can be used as a starting point for your own workloads.

## ⟦C1⟧ — topic to read from; {#topic-read}

Data is read from a topic 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. This pattern is used in all subsequent examples.

{% note info %}

Working with [local and external topics](https://ydb.tech/docs/en/dev/streaming-query/local-and-external-topics.md?version=main) is shown.

In the examples:

- `ext_source` — a pre-created `external data source`.
- `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.

{% endnote %}

The following fragment shows reading events from a topic in JSON format. It is used inside [CREATE STREAMING QUERY](https://ydb.tech/docs/en/yql/reference/syntax/create-streaming-query.md?version=main) in the `DO BEGIN ... END DO` block:


```yql
SELECT
    *
FROM
    ext_source.input_topic -- or local topic input_topic
WITH (
    FORMAT = json_each_row,
    SCHEMA = (
        Id Uint64 NOT NULL,
        Name Utf8 NOT NULL
    )
);
```


The following snippet reads JSON events from a topic. Use it inside [CREATE STREAMING QUERY](https://ydb.tech/docs/en/dev/streaming-query/streaming-query-formats.md?version=main) in a ⟦C1⟧ block:

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

The query reads events from the input topic, builds a JSON object from fields, and writes to the output topic. `AsStruct` builds a structure from the fields, `Yson::From` converts it to Yson, `Yson::SerializeJson` serializes to a JSON string, and `ToBytes` converts to `String`, which is required for topic writes.


```yql
CREATE STREAMING QUERY write_json_example AS
DO BEGIN

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


More on the functions:

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

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

The query reads events from the input topic and writes a single field as a string to the output topic. Topic writes require `SELECT` to 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
```


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

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

The query reads events from a topic and writes them to `output_table`. Create the table beforehand with a schema that matches the selected columns.

{% note warning %}

Table writes in streaming queries support **UPSERT only**. `INSERT INTO` is not supported: with [at-least-once](https://ydb.tech/docs/en/concepts/streaming-query/streaming-query.md?version=main#guarantees) delivery, retries would duplicate rows. With `UPSERT`, an existing row with the same primary key is updated; otherwise a new row is inserted, while `INSERT INTO` fails.

{% 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
```


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

## See also

- [Local and external topics in streaming queries](https://ydb.tech/docs/en/dev/streaming-query/local-and-external-topics.md?version=main)
- [CREATE STREAMING QUERY](https://ydb.tech/docs/en/yql/reference/syntax/create-streaming-query.md?version=main)
- [Quick start: reading and writing to topics](https://ydb.tech/docs/en/recipes/streaming_queries/topics.md?version=main)