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

# Writing to tables

Writing to tables lets you persist streaming query results for analysis with regular SQL. For example, you can aggregate events from a stream and store summaries in a table.

Writes use [UPSERT INTO](https://ydb.tech/docs/en/yql/reference/syntax/upsert_into.md?version=v25.4) — insert a new row or update an existing row by primary key. UPSERT is idempotent by primary key: writing the same row again updates it rather than duplicating. That matters because streaming queries provide [at-least-once](https://ydb.tech/docs/en/concepts/streaming-query.md?version=v25.4#guarantees) delivery — after recovery from a [checkpoint](https://ydb.tech/docs/en/dev/streaming-query/checkpoints.md?version=v25.4), some events may be processed more than once.

{% note alert %}

Not supported:

- [INSERT INTO](https://ydb.tech/docs/en/yql/reference/syntax/insert_into.md?version=v25.4) — use UPSERT INTO instead. `INSERT INTO` would duplicate rows on retries under at-least-once delivery.
- Writing to YDB tables in **external** databases. Currently only local tables can be written to.

{% endnote %}

## Example

The query reads events from a topic and writes them to `output_table`. `Ts` is cast from string to `Timestamp`, and [Unwrap](https://ydb.tech/docs/en/yql/reference/builtins/basic.md?version=v25.4#unwrap) removes optionality.

```sql
CREATE STREAMING QUERY query_with_table_write AS
DO BEGIN

-- Read from topic and write to table
UPSERT INTO
    output_table
SELECT
    -- String to Timestamp
    Unwrap(CAST(Ts AS Timestamp)) AS Ts,
    Country,
    Count
FROM
    -- Read events from topic
    ydb_source.input_topic
WITH (
    -- Topic data format
    FORMAT = json_each_row,
    -- Data schema
    SCHEMA = (
        Ts String NOT NULL,
        Count Uint64 NOT NULL,
        Country Utf8 NOT NULL
    )
);

END DO
```