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

# Quick start: reading and writing to topics

In this guide, you will create your first [streaming query](https://ydb.tech/docs/en/concepts/streaming-query/streaming-query.md?version=main).

The query will:

- read events from the input [topic](https://ydb.tech/docs/en/concepts/datamodel/topic.md?version=main)
- filter only errors
- count the number of errors per server over 10 minutes
- write the result to the output topic.

Events arrive in JSON format with fields: time, logging level, and server name.

You will perform the following steps:

* [Create topics](#step1);
* [Create the streaming query](#step2);
* [Check query state](#step3);
* [Produce sample input](#step4);
* [Read the output topic](#step5);
* [Delete the streaming query](#step6).

## Prerequisites {#requirements}

To run the examples, you will need:

* A running YDB database — see [quick start](https://ydb.tech/docs/en/quickstart.md?version=main);
* The `enable_streaming_queries` feature flag enabled.

{% list tabs %}

- Docker

  ```bash
  docker run -d --rm --name ydb-local -h localhost \
    --platform linux/amd64 \
    -p 2135:2135 -p 2136:2136 -p 8765:8765 -p 9092:9092 \
    -v $(pwd)/ydb_certs:/ydb_certs \
    -e GRPC_TLS_PORT=2135 -e GRPC_PORT=2136 -e MON_PORT=8765 \
    -e YDB_FEATURE_FLAGS=enable_streaming_queries \
    ydbplatform/local-ydb:25.4
  ```

- local_ydb

  ```bash
  ./local_ydb deploy \
    --ydb-working-dir=/absolute/path/to/working/directory \
    --ydb-binary-path=/path/to/kikimr/driver \
    --enable-feature-flag=enable_streaming_queries
  ```

{% endlist %}

<!-- source: en/_includes/ydb-cli-profile.md -->
{% note info %}

The examples use the `quickstart` profile. To learn more, see [Creating a profile to connect to a test database](https://ydb.tech/docs/en/reference/ydb-cli/profile/create.md?version=main#quickstart).

{% endnote %}
<!-- endsource: en/_includes/ydb-cli-profile.md -->

## Step 1. Creating topics {#step1}

Create the input and output [topics](https://ydb.tech/docs/en/concepts/datamodel/topic.md?version=main):


```sql
CREATE TOPIC input_topic;
CREATE TOPIC output_topic;
```


Verify that the topics are created:


```bash
./ydb --profile quickstart scheme ls
```


## Step 2. Create the streaming query {#step2}

Create a [streaming query](https://ydb.tech/docs/en/concepts/streaming-query/streaming-query.md?version=main) with [CREATE STREAMING QUERY](https://ydb.tech/docs/en/yql/reference/syntax/create-streaming-query.md?version=main):


```sql
CREATE STREAMING QUERY query_example AS
DO BEGIN

$number_errors = SELECT
    Host,
    COUNT(*) AS ErrorCount,
    CAST(HOP_START() AS String) AS Ts  -- Start time of the window corresponding to the aggregation result
FROM
    input_topic
WITH (
    FORMAT = json_each_row,
    SCHEMA = (
        Time String NOT NULL,
        Level String NOT NULL,
        Host String NOT NULL
    )
)
WHERE
    Level = "error"
GROUP BY
    HOP(CAST(Time AS Timestamp), "PT600S", "PT600S", "PT0S"),  -- Number of errors on non-overlapping windows of 10 minutes length
    Host;

INSERT INTO
    output_topic
SELECT
    ToBytes(Unwrap(Yson::SerializeJson(Yson::From(TableRow()))))  -- Serialization of all columns into JSON
FROM
    $number_errors;

END DO
```


More details:

- `GROUP BY HOP` aggregation and `HOP_START` — [GROUP BY ... HOP](https://ydb.tech/docs/en/yql/reference/syntax/select/group-by.md?version=main#group-by-hop).
- Writing to a topic — [Formats for writing data](https://ydb.tech/docs/en/dev/streaming-query/streaming-query-formats.md?version=main#write_formats).
- JSON serialization: [TableRow](https://ydb.tech/docs/en/yql/reference/builtins/basic.md?version=main#tablerow), [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).

## Step 3. Check query state {#step3}

Check query state in the [streaming_queries](https://ydb.tech/docs/en/dev/system-views.md?version=main#streaming_queries) system table:


```sql
SELECT
    Path,
    Status,
    Issues,
    Run
FROM
    `.sys/streaming_queries`
```


Make sure that the `Status` field has the value `RUNNING`. Otherwise, check the `Issues` field.

If the query is in `SUSPENDED` status or the `Issues` field contains errors, see the error diagnostics section.

## Step 4. Produce sample input {#step4}

Write test messages to the topic using the [YDB CLI](https://ydb.tech/docs/en/reference/ydb-cli/index.md?version=main):


```bash
echo '{"Time": "2025-01-01T00:00:00.000000Z", "Level": "error", "Host": "host-1"}' | ./ydb --profile quickstart topic write input_topic
echo '{"Time": "2025-01-01T00:04:00.000000Z", "Level": "error", "Host": "host-2"}' | ./ydb --profile quickstart topic write input_topic
echo '{"Time": "2025-01-01T00:08:00.000000Z", "Level": "error", "Host": "host-1"}' | ./ydb --profile quickstart topic write input_topic
echo '{"Time": "2025-01-01T00:12:00.000000Z", "Level": "error", "Host": "host-2"}' | ./ydb --profile quickstart topic write input_topic
echo '{"Time": "2025-01-01T00:12:00.000000Z", "Level": "error", "Host": "host-1"}' | ./ydb --profile quickstart topic write input_topic
```


Results appear in the output topic after the 10-minute aggregation window closes.

## Step 5. Read the output topic {#step5}

Read data from the output topic:


```bash
./ydb --profile quickstart topic read output_topic --partition-ids 0 --start-offset 0 --limit 10 --format newline-delimited
```


Expected result:


```json
{"ErrorCount":1,"Host":"host-2","Ts":"2025-01-01T00:00:00Z"}
{"ErrorCount":2,"Host":"host-1","Ts":"2025-01-01T00:00:00Z"}
```


## Step 6. Delete the query {#step6}

Delete the query with [DROP STREAMING QUERY](https://ydb.tech/docs/en/yql/reference/syntax/drop-streaming-query.md?version=main):


```sql
DROP STREAMING QUERY query_example;
```


## What's next {#next-steps}

- Explore the [data formats](https://ydb.tech/docs/en/dev/streaming-query/streaming-query-formats.md?version=main) supported in streaming queries.
- Learn how to [enrich data with a reference](https://ydb.tech/docs/en/dev/streaming-query/enrichment.md?version=main) from a local table or from S3.
- Learn how to [write results to tables](https://ydb.tech/docs/en/dev/streaming-query/table-writing.md?version=main).

## See also

* [Streaming queries](https://ydb.tech/docs/en/concepts/streaming-query/streaming-query.md?version=main)
* [Data formats for reading/writing from topics](https://ydb.tech/docs/en/dev/streaming-query/streaming-query-formats.md?version=main).