[Yandex Cloud documentation](../../index.md) > [Yandex Query](../index.md) > Data sources and sinks > Working with Data Streams > Writing data

# Writing data to Yandex Data Streams

[Yandex Data Streams](../../data-streams/concepts/index.md) 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

```sql
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 connection {#connect}

To read data from Yandex Data Streams:
1. [Navigate](../../console/operations/select-service.md#select-service) to the **Connections** section of the **Yandex Query** interface and click **Create new**.
1. In the window that opens, specify the Yandex Data Streams connection name in the **Name** field.
1. In the **Type** dropdown, select `Data Streams`.
1. In the **Database** dropdown, select the Yandex Managed Service for YDB database where you created the Yandex Data Streams stream.
1. In the **Service account** field, select an existing service account or create a new one. Assign it the [`yds.writer`](../../data-streams/security/index.md) permissions required to read data.
1. Click **Create** to create a connection.

## Data model

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

```sql
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 example

Query example for reading data from Yandex Data Streams and writing the results to Yandex Data Streams:

```sql
$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:

|Field|Type|Description|
|--|---|---|
|`yds`| |Yandex Data Streams connection name|
|`input_stream`| |Name of the source data stream in the SQL query|
|`output_stream`| |Target stream name in the SQL query|
|`host`|String|String query parameter|
|`count`|Integer|Numeric query parameter|
|`raw`|String|Data format. Support is currently limited to the `raw` format|

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

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

The YQL guides provide details on [Yson](https://ydb.tech/docs/en//yql/reference/udf/list/yson) and [Json](https://ydb.tech/docs/en//yql/reference/types/json) modules, and [their functions](https://ydb.tech/docs/en//yql/reference/builtins/json), [<|"key": value|>](https://ydb.tech/docs/en//yql/reference/builtins/struct).

## Supported 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.