Reading data from Data Streams via Query connections

Using Yandex Data Streams connections is convenient for prototyping and initial data access configuration.

Yandex Data Streams enables you to transfer data streams to multiple applications for processing, with each handling the data independently of the others.

Query example for reading Json-formatted data from 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;

Setting up a connectionSetting up a connection

To read data from Yandex Data Streams:

1. Navigate to the Connections section of the Yandex Query interface and click Create new.

2. In the window that opens, specify the Yandex Data Streams connection name in the Name field.

3. In the Type dropdown, select Data Streams.

4. In the Cloud and Folder field, specify the data source location.

5. In the Database dropdown, select the Yandex Managed Service for YDB database where you created the Yandex Data Streams stream.

6. In the Service account field, select an existing service account or create a new one. Assign it the yds.editor permissions required to read data.

   To use a service account, the iam.serviceAccounts.user role is required.

7. Click Create to create a connection.

Data modelData model

Data is transmitted via Yandex Data Streams in binary format and is read via SQL statements.

SELECT
    <expression>
FROM
    <connection>.<stream_name>
WITH
(
    format=raw,
    SCHEMA
    (
        Data String
    )
)
WHERE <filter>;

Where:

• <connection>: Name of the Data Streams data stream connection created in the previous step.
• <stream_name>: Data Streams data stream name.

Data reading exampleData reading example

Query example for reading data from Yandex Data Streams and writing the results to 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;

Where: