---
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: ru/core/recipes/streaming_queries/topics.md
---
> **Documentation Index:** Fetch the complete configuration index at https://ydb.tech/docs/ru/llms.txt

# Быстрый старт: чтение и запись в топики

В этом руководстве вы создадите свой первый [потоковый запрос](https://ydb.tech/docs/ru/concepts/streaming-query/streaming-query.md?version=main).

Запрос будет:

- читать события из входного [топика](https://ydb.tech/docs/ru/concepts/datamodel/topic.md?version=main);
- отбирать только ошибки;
- подсчитывать количество ошибок по каждому серверу за 10 минут;
- записывать результат в выходной топик.

События поступают в формате JSON с полями: время, уровень логирования и имя сервера.

Вы выполните следующие шаги:

* [создание топиков](#step1);
* [создание потокового запроса](#step2);
* [просмотр состояния запроса](#step3);
* [заполнение входного топика данными](#step4);
* [проверка содержимого выходного топика](#step5);
* [удаление потокового запроса](#step6).

## Предварительные условия {#requirements}

Для выполнения примеров вам потребуется:

* запущенная база YDB — см. [quick start](https://ydb.tech/docs/ru/quickstart.md?version=main);
* включённый флаг `enable_streaming_queries`.

{% 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: ru/_includes/ydb-cli-profile.md -->
{% note info %}

В примерах используется профиль `quickstart`, подробнее смотрите в [Создание профиля для соединения с тестовой БД](https://ydb.tech/docs/ru/reference/ydb-cli/profile/create.md?version=main#quickstart).

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

## Шаг 1. Создание топиков {#step1}

Создайте входной и выходной [топики](https://ydb.tech/docs/ru/concepts/datamodel/topic.md?version=main):

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

Проверьте, что топики созданы:

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

## Шаг 2. Создание потокового запроса {#step2}

Создайте [потоковый запрос](https://ydb.tech/docs/ru/concepts/streaming-query/streaming-query.md?version=main) с помощью [CREATE STREAMING QUERY](https://ydb.tech/docs/ru/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  -- Время начала окна, соответствующего результату агрегации
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"),  -- Число ошибок на неперекрывающихся окнах длиной 10 минут
    Host;

INSERT INTO
    output_topic
SELECT
    ToBytes(Unwrap(Yson::SerializeJson(Yson::From(TableRow()))))  -- Сериализация всех колонок в JSON
FROM
    $number_errors;

END DO
```

Подробнее:

- Агрегация `GROUP BY HOP` и функция `HOP_START` — [GROUP BY ... HOP](https://ydb.tech/docs/ru/yql/reference/syntax/select/group-by.md?version=main#group-by-hop).
- Запись данных в топик — [Форматы при записи данных](https://ydb.tech/docs/ru/dev/streaming-query/streaming-query-formats.md?version=main#write_formats).
- Сериализация в JSON: [TableRow](https://ydb.tech/docs/ru/yql/reference/builtins/basic.md?version=main#tablerow), [Yson::From](https://ydb.tech/docs/ru/yql/reference/udf/list/yson.md?version=main#ysonfrom), [Yson::SerializeJson](https://ydb.tech/docs/ru/yql/reference/udf/list/yson.md?version=main#ysonserializejson), [Unwrap](https://ydb.tech/docs/ru/yql/reference/builtins/basic.md?version=main#unwrap), [ToBytes](https://ydb.tech/docs/ru/yql/reference/builtins/basic.md?version=main#to-from-bytes).

## Шаг 3. Просмотр состояния запроса {#step3}

Проверьте состояние запроса через системную таблицу [streaming_queries](https://ydb.tech/docs/ru/dev/system-views.md?version=main#streaming_queries):

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

Убедитесь, что в поле `Status` значение `RUNNING`. В противном случае проверьте поле `Issues`.

Если запрос находится в статусе `SUSPENDED` или в поле `Issues` есть ошибки, обратитесь к разделу диагностика ошибок.


## Шаг 4. Заполнение входного топика данными {#step4}

Запишите тестовые сообщения в топик с помощью [YDB CLI](https://ydb.tech/docs/ru/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
```

Результат появится в выходном топике после закрытия 10-минутного окна агрегации.

## Шаг 5. Проверка содержимого выходного топика {#step5}

Прочитайте данные из выходного топика:

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

Ожидаемый результат:

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

## Шаг 6. Удаление запроса {#step6}

Удалите запрос с помощью [DROP STREAMING QUERY](https://ydb.tech/docs/ru/yql/reference/syntax/drop-streaming-query.md?version=main):

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

## Что дальше {#next-steps}

- Изучите [форматы данных](https://ydb.tech/docs/ru/dev/streaming-query/streaming-query-formats.md?version=main), поддерживаемые в потоковых запросах.
- Узнайте, как [обогащать данные справочником](https://ydb.tech/docs/ru/dev/streaming-query/enrichment.md?version=main) из локальной таблицы или из S3.
- Научитесь [записывать результаты в таблицы](https://ydb.tech/docs/ru/dev/streaming-query/table-writing.md?version=main).

## См. также

* [Потоковые запросы](https://ydb.tech/docs/ru/concepts/streaming-query/streaming-query.md?version=main);
* [Форматы данных при чтении/записи из топиков](https://ydb.tech/docs/ru/dev/streaming-query/streaming-query-formats.md?version=main).