[Yandex Cloud documentation](../../../../index.md) > [Yandex Data Transfer](../../../index.md) > [Step-by-step guides](../../index.md) > [Configuring endpoints](../index.md) > Yandex Data Streams > Target

# Transferring data to a Yandex Data Streams target endpoint


Yandex Data Transfer enables you to migrate data to a Yandex Data Streams queue and implement various data processing and transformation scenarios. To implement a transfer:

1. [Explore possible data transfer scenarios](#scenarios).
1. [Configure one of the supported data sources](#supported-sources).
1. [Configure the target endpoint](#endpoint-settings) in Yandex Data Transfer.
1. [Create](../../transfer.md#create) a transfer and [start](../../transfer.md#activate) it.
1. Perform required operations with the database and [control the transfer](../../monitoring.md).
1. In case of any issues, [use ready-made solutions](#troubleshooting) to resolve them.

## Scenarios for transferring data to Yandex Data Streams {#scenarios}

1. Data change capture means tracking changes to a database and delivering those changes to consumers. It is used for applications that are sensitive to real-time data changes.
    * [MySQL® change data capture and delivery to YDS](../../../tutorials/mmy-to-yds.md).
    * [PostgreSQL change data capture and delivery to YDS](../../../tutorials/mpg-to-yds.md).
    * [YDB change data capture and delivery to YDS](../../../tutorials/ydb-to-yds.md).

1. Data delivery is a process of delivering arbitrary data to target storage. It includes data retrieval from a queue and its deserialization with subsequent transformation to target storage format.
  
    * [Delivering data from Apache Kafka® to YDS](../../../tutorials/mkf-to-yds.md).

1. Migration: Moving data from one storage to another. Migration often means migrating a database from obsolete local databases to managed cloud ones.

For a detailed description of possible Yandex Data Transfer scenarios, see [Tutorials](../../../tutorials/index.md).

## Configuring the data source {#supported-sources}

Configure one of the supported data sources:

* [PostgreSQL](../source/postgresql.md)
* [MySQL®](../source/mysql.md)
* [Apache Kafka®](../source/kafka.md)
* [Managed Service for YDB](../source/ydb.md)
* [Airbyte®](../../../transfer-matrix.md#airbyte)
* [YDS](../source/data-streams.md)
* [Yandex Object Storage](../source/object-storage.md)
* [OpenSearch](../source/opensearch.md)

For a complete list of supported sources and targets in Yandex Data Transfer, see [Available transfers](../../../transfer-matrix.md).

## Configuring the Yandex Data Streams target endpoint {#endpoint-settings}

When [creating](../index.md#create) or [updating](../index.md#update) an endpoint, you can define:

* Stream [connection settings](#general-settings) in Yandex Data Streams and [serialization settings](#serializer). These are required parameters.
* [Advanced settings](#additional-settings).

### Basic settings {#general-settings}

{% list tabs group=instructions %}

- Management console {#console}

    * **Database**: Select a Yandex Managed Service for YDB database registered in [Yandex Data Streams](../../../../data-streams/index.md) as a target.
    
    * **Stream**: Specify the name of the data stream associated with the database.
    
    * **Service account**: Select or create a [service account](../../../../iam/concepts/users/service-accounts.md) with the `yds.editor` role that Data Transfer will use to connect to the data target.
    
    * **Security groups**: Select the cloud network to host the endpoint and security groups for network traffic.
    
      Thus, you will be able to apply the specified security group rules to the VMs and clusters in the selected network without changing the settings of these VMs and clusters. For more information, see [Networking in Yandex Data Transfer](../../../concepts/network.md).

{% endlist %}

### Advanced settings {#additional-settings}

{% list tabs group=instructions %}

- Management console {#console}

    * **Save transaction order**: Do not split an event stream into independent queues by table.

    * **Compresion codecs**: Set the required data compression format (`GZIP`, `ZSTD`, or `Raw`).

{% endlist %}

### Serializing settings {#serializer}

{% list tabs group=instructions %}

- Management console {#console}

    * Under **Serializing settings**, select the [serialization](../../../concepts/serializer.md) type:
    
      * **Auto**: Automatic serialization.
      * **Debezium**: Serialization under the Debezium standards:
      
        * **Yandex Schema Registry**:
    
          * Select a message schema (data format).
          * From the list, select the [namespace](../../../../metadata-hub/operations/list-name-space.md) of the Yandex Schema Registry schema registry.
    
        * **On Premise**:
    
          * Select the message key schema (matches the `key.converter` Debezium parameter). The default value matches the `key.converter.schemas.enable=true` and `key.converter=org.apache.kafka.connect.json.JsonConverter` Debezium parameters. If you select **Disabled**, the `key.converter.schemas.enable` Debezium parameter takes the `false` value.
          * Select the message value schema (matches the `value.converter` Debezium parameter). The default value matches the `value.converter.schemas.enable=true` and `value.converter=org.apache.kafka.connect.json.JsonConverter` Debezium parameters. If you select **Disabled**, the `value.converter.schemas.enable` Debezium parameter takes the `false` value.
          * When selecting the **Avro (via Schema Registry)**, **JSON (via Schema Registry)**, or **Protobuf (via Schema Registry)** value, specify the parameters for access to Confluent Schema Registry for message key and value schemas:
    
            * **URL**: URL.
            * **Username**. This is an optional setting.
            * **Password**. This is an optional setting.
            * **CA certificate**: Upload the SSL certificate for the connection. This is an optional setting. To skip the certificate check, do not provide one.
    
        * If required, specify [**Debezium serializer settings**](../../../concepts/serializer.md#debezium) in `Parameter`-`Value` format.
    
    If you want to use JSON schemas in Yandex Schema Registry and preserve their [compatibility when adding and deleting optional fields](../../../../metadata-hub/concepts/schema-registry-content-model.md#optional-parameters-compatibility-solution), use these settings:
    
    **Serializing settings**: **Debezium**: **On Premise**.
    
    * To use Schema Registry for keys, select **Key schema type**: **JSON (via Schema Registry)**. To use Schema Registry for values, select **Value schema type**: **JSON (via Schema Registry)**. Set the parameters for access to Schema Registry:
      * **URL**: `https://<namespace_ID>.schema-registry.yandexcloud.net:443`.
    
          {% note warning %}
    
          The namespace must have **Compatibility check policy for JSON** [set](../../../../metadata-hub/operations/update-name-space.md) to `optional friendly`.
    
          {% endnote %}
    
      * **Username**: `api-key`.
    
      
      * **Password**: Value of the [API key](../../../../iam/concepts/authorization/api-key.md) with a limited scope used for connecting to Schema Registry. To get this value:
          1. Create an API key with a limited scope and place it in the `SECRET` local variable:
    
              ```bash
              yc iam api-key create --folder-id <folder_ID> \
                --service-account-name <name_of_service_account_for_operations_with_Schema_Registry> \
                --scopes yc.schema-registry.schemas.manage \
                --expires-at '2030-01-01T00:00:00Z' >./api-key.yaml && \
              SECRET=`cat ./api-key.yaml | grep 'secret:' | awk '{print $2}'`
              ```
    
          1. Print the `SECRET` variable value to the terminal:
    
              ```bash
              echo $SECRET
              ```
    
    
    * Under **Debezium serializer settings**:
        * To generate a closed schema for keys, include the `key.converter.dt.json.generate.closed.content.schema` parameter set to `true`.
        * To generate a closed schema for values, include the `value.converter.dt.json.generate.closed.content.schema` parameter set to `true`.

{% endlist %}

After configuring the data source and target, [create and start the transfer](../../transfer.md#create).

## Troubleshooting data transfer issues {#troubleshooting}

* [Transfer interrupted with an error message](#overloaded)
* [Cloud Functions redirects](#redirects)

For more troubleshooting tips, see [Troubleshooting](../../../troubleshooting/index.md).

### Transfer failure {#overloaded}

A _**Replication**_ or _**Snapshot and increment**_ transfer is interrupted with an error.

Error message:

```text
/Ydb.PersQueue.V1.PersQueueService/AddReadRule failed: OVERLOADED
```

Transfers are aborted due to the cloud [quota](https://console.yandex.cloud/cloud?section=quotas) on the number of operations with Managed Service for YDB.

**Solution:**

1. Increase the **Number of schema transactions per minute** property in the [Managed Service for YDB quotas](../../../../ydb/concepts/limits.md) for the cloud with the required database and [reactivate](../../transfer.md#activate) the transfer.

### Cloud Functions redirects {#redirects}

In rare cases, the following error may occur during transfers from Data Streams or Apache Kafka®:

```text
redirect to SOME_URL is requested but no redirects are allowed.
```

Possible cause:

The use of the Cloud Functions function is set up on the source. It returns a redirect to another URL rather than data.

**Solution:**

Such redirects are not allowed for security reasons. Avoid using redirects to Cloud Functions during transfers.