Чтение данных из Data Streams с помощью соединений в Query

Соединения удобно использовать для прототипирования и первоначальной настройки подключения к данным Yandex Data Streams.

Yandex Data Streams — сервис для передачи потоков данных нескольким приложениям. Каждое приложение обрабатывает данные независимо от других.

Пример чтения данных в формате JSON из Yandex Data Streams:

SELECT
    JSON_VALUE(CAST(Data AS Json), "$.action") AS action
FROM yds.`input_stream`
WITH (
    format=raw,
    SCHEMA
    (
        Data String
    )
)
LIMIT 10;

Настройка соединенияНастройка соединения

Чтобы настроить чтение данных из Yandex Data Streams:

1. Перейдите в сервис Yandex Query.

2. На панели слева выберите Соединения.

3. Нажмите Создать.

4. В открывшемся окне в поле Имя укажите название соединения с Yandex Data Streams.

5. В выпадающем поле Тип выберите Data Streams.

6. В поле Облако и каталог выберите расположение источника данных.

7. В выпадающем поле База данных выберите базу данных Yandex Managed Service for YDB, где ранее был создан поток Yandex Data Streams.

8. В поле Сервисный аккаунт выберите сервисный аккаунт, который будет использоваться для чтения данных, или создайте новый, выдав ему права yds.editor.

   Чтобы использовать сервисный аккаунт, пользователю нужна роль iam.serviceAccounts.user.

9. Нажмите Создать.

Модель данныхМодель данных

Данные через Yandex Data Streams передаются в бинарном виде. Для чтения данных используйте SQL-выражение следующего вида:

SELECT
    <выражение>
FROM
    <соединение>.<имя_потока>
WITH
(
    format=raw,
    SCHEMA
    (
        Data String
    )
)
WHERE
    <фильтр>;

Где:

• <выражение> — выражение, определяющее результат запроса;
• <соединение> — название соединения с потоком данных Data Streams, созданного в предыдущем разделе;
• <имя_потока> — название потока данных в Data Streams;
• <фильтр> — условие фильтрации данных.

Пример чтения данныхПример чтения данных

Пример запроса для чтения данных из Yandex Data Streams:

$data =
SELECT
    JSON_VALUE(Data, "$.host") AS host,
    JSON_VALUE(Data, "$.count") AS count,
    JSON_VALUE(Data, "$.tag") AS tag,
FROM
(
    SELECT
        CAST(Data AS Json) AS Data
    FROM yds.`input_stream`
    WITH
    (
        format=raw,
        SCHEMA
        (
            Data String
        )
    )
)
WHERE
    JSON_VALUE(Data, "$.tag") = "my_tag";

SELECT
    *
FROM
    $data
LIMIT 10;

Где: