Writing data from Yandex Query to Yandex Data Streams streams

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

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 set up writing data to 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 field, select Data Streams.
6. In the Database field, select the Yandex Managed Service for YDB database where you created the Yandex Data Streams stream.
7. In the Service account field, select the service account you intend to use to write data, or create a new one and assign it the yds.writer role.
8. Click Create.

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 section.
• <stream_name>: Data Streams data stream name.
• <statement>: Statement defining the data to write.
• <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. For simpler processing, the results are converted to JSON format using the following structure:

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

See YQL guides for detailed descriptions of the Yson and Json modules, JSON functions, and structures.

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.