Writing data to Yandex Data Streams

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 writing JSON-formatted data to Yandex Data Streams

INSERT INTO yds.`output_stream`
SELECT
    ToBytes(Unwrap(Json::SerializeJson(Yson::From(
    <|"predefined":
            <|
                "host": host,
                "count": count,
            |>,
            "optional":
            <|
                "tag": tag
            |>
        |>))))
FROM
    $data;

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 Database dropdown, select the Yandex Managed Service for YDB database where you created the Yandex Data Streams stream.
5. In the Service account field, select an existing service account or create a new one. Assign it the yds.writer permissions required to read data.
6. Click Create to create a connection.

Data modelData model

Data is transmitted via Yandex Data Streams in binary format and is written via SQL statements as follows:

INSERT INTO <connection>.<stream_name>
    <expression>
FROM
   <query>

Where:

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

Data writing exampleData writing 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,
    CAST(JSON_VALUE(Data, "$.count") AS Int) 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";

INSERT INTO yds.`output_stream`
SELECT
    ToBytes(Unwrap(Json::SerializeJson(Yson::From(
    <|"predefined":
            <|
                "host": host,
                "count": count,
            |>,
            "optional":
            <|
                "tag": tag
            |>
        |>))))
FROM
    $data;

Where:

The system writes processing results to the Yandex Data Streams output stream. To facilitate further processing, the data is converted to JSON as follows:

    ToBytes(Unwrap(Json::SerializeJson(Yson::From(
    <|"key": value|>,
    <|"key2":
        <|"child_key": child_value|>,
    |>,
    ))))

The YQL guides provide details on Yson and Json modules, and their functions, <|"key": value|>.

Supported write formatsSupported write formats

Data Streams only lets you write data as a byte stream, which is processed by the receiving side.

When writing to Data Streams, file format and compression algorithm settings are not applied.