Reading data from Data Streams via Query connections

Connections are useful for prototyping and initial setup of Yandex Data Streams data connections.

Yandex Data Streams: Service for transmitting data streams to multiple applications. Each application processes data independently from the others.

Example of reading JSON 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 set up reading from Yandex Data Streams:

1. Navigate to Yandex Query.

2. In the left-hand panel, select Connections.

3. Click Create new.

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

5. In the Type dropdown, select Data Streams.

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

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

8. 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 the service account on your behalf, you need the iam.serviceAccounts.user role.

9. Click Create.

Data modelData model

Data is transmitted via Yandex Data Streams in binary format. To read data, use an SQL statement of the following format:

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

Where:

• <statement>: Statement that determines the result of the query.
• <connection>: Name of the Data Streams data stream connection created in the previous section.
• <stream_name>: Data Streams data stream name.
• <filter>: Data filtering condition.

Data reading exampleData reading example

Query example for reading data from 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: