Writing data to Yandex Data Streams

Yandex Data Streams is a service that allows you to transfer data streams to multiple applications for processing, with each of them handling the data independently.

Example of writing JSON 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. Go to Connections in the Yandex Query interface and click Create new.
2. In the window that opens, specify a name for a connection to Yandex Data Streams in the Name field.
3. In the drop-down list under Type, select Data Streams.
4. In the Database drop-down list, select the Yandex Managed Service for YDB database where the Yandex Data Streams stream was created.
5. In the Service account field, select a service account for data reads or create a new one with the yds.writer permissions.
6. Click Create to create a connection.

Data modelData model

Data is sent via Yandex Data Streams in binary form. Data is written using SQL statements and generally looks like this:

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>: Name of the data stream in Data Streams.
• <query>: Yandex Query data source query.

Example of writing dataExample of writing data

Sample query 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 processing results are written to the Yandex Data Streams output stream. To facilitate the processing, this data is converted to JSON format using the following statement:

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

The YQL documentation provides a detailed description of Yson, Json and its functions, <|"key": value|>.

Supported write formatsSupported write formats

Data Streams only lets you write data as a byte stream that is interpreted on the receiving side.

File format and compression algorithm settings for data writes in Data Streams are not applied.