Contact UsTry it for free
© 2026 Direct Cursus Technology L.L.C.

Transferring data to Yandex Managed Service for Apache Kafka® using Yandex Data Transfer

Written by
Updated at January 22, 2026

You can track data changes in a Managed Service for MySQL® source cluster and send them to a Managed Service for Apache Kafka® target cluster using change data capture (CDC).

To set up CDC using Data Transfer:

  1. Set up the source cluster.
  2. Set up the target cluster.
  3. Set up and activate the transfer.
  4. Test your transfer.

If you no longer need the resources you created, delete them.

Getting started

  1. Create a Managed Service for MySQL® source cluster in any suitable configuration with the following settings:

    • Database: db1
    • User: my-user
    • Hosts: Publicly available

  2. Create a Managed Service for Apache Kafka® target cluster in any suitable configuration with publicly available hosts.

  3. If using security groups, configure them to allow internet access to your clusters:

  4. Install kcat (formerly known as kafkacat) and the MySQL command-line tool on your local machine. For example, if your computer is running Ubuntu 20.04, use this command:

    sudo apt update && sudo apt install kafkacat mysql-client --yes

    Make sure you can use it to connect to the Managed Service for Apache Kafka® source cluster over SSL.

Set up the source cluster

  1. For Data Transfer to get notifications of data changes from a Managed Service for MySQL® cluster, configure external replication in the source cluster. To enable my-user to perform replication, assign the ALL_PRIVILEGES role for the db1 database and grant the REPLICATION CLIENT and REPLICATION SLAVE global privileges to the user.

  2. Connect to the db1 database as my-user.

  3. Add test data to the database. In this example, we will use a simple table containing information from certain car sensors.

    Create a table:

    CREATE TABLE db1.measurements (
    device_id varchar(200) NOT NULL,
    datetime timestamp NOT NULL,
    latitude real NOT NULL,
    longitude real NOT NULL,
    altitude real NOT NULL,
    speed real NOT NULL,
    battery_voltage real,
    cabin_temperature real NOT NULL,
    fuel_level real,
    PRIMARY KEY (device_id)
);

    Populate the table with data:

    INSERT INTO db1.measurements VALUES
    ('iv9a94th6rzt********', '2022-06-05 17:27:00', 55.70329032, 37.65472196,  427.5,    0, 23.5, 17, NULL),
    ('rhibbh3y08qm********', '2022-06-06 09:49:54', 55.71294467, 37.66542005, 429.13, 55.5, NULL, 18, 32);

Set up the target cluster

The settings vary depending on the topic management method used. Data topic names follow the <topic_prefix>.<schema_name>.<table_name> convention. In this tutorial, we will use the cdc prefix.

When managing topics using native Yandex Cloud interfaces (management console, CLI, Terraform, or API):

  1. Create a topic named cdc.db1.measurements.

    To track changes in multiple tables, create a separate topic for each one.

  2. Create a user named kafka-user with ACCESS_ROLE_CONSUMER and ACCESS_ROLE_PRODUCER roles for the new topics. To include all such topics, use cdc.* as the topic name.

When managing topics using the Kafka Admin API:

  1. Create an admin user named kafka-user.

  2. In addition to ACCESS_ROLE_ADMIN, assign this user the ACCESS_ROLE_CONSUMER and ACCESS_ROLE_PRODUCER roles for all topics prefixed with cdc.

    Required topics will be created automatically upon the first change to the source cluster tables you are tracking. This solution can be useful to track changes in multiple tables but requires extra free space in the cluster storage. For more information, see Storage in Managed Service for Apache Kafka®.

Set up and activate the transfer

  1. Create a MySQL® source endpoint with the following settings:

    • Database type: MySQL.
    • Endpoint parameters:
      • Connection settings: Managed Service for MySQL cluster.
      • Managed Service for MySQL cluster: Select the Managed Service for MySQL® cluster you created earlier.
      • Database: db1.
      • User: my-user.
      • Password: Enter the my-user password.
      • Included tables: db1.measurements.

  2. Create an Apache Kafka® target endpoint with the following settings:

    • Database type: Kafka.

    • Endpoint parameters:

      • Connection type: Managed Service for Apache Kafka cluster.

        • Managed Service for Apache Kafka cluster: Select the Managed Service for Apache Kafka® cluster you created earlier.
        • Authentication: Specify the kafka-user credentials.

      • Topic: Topic full name.

      • Topic full name: cdc.db1.measurements.

      If you need to track changes in multiple tables, fill out the fields as follows:

      • Topic: Topic prefix.
      • Topic prefix: Enter the cdc prefix you used to generate topic names.

  3. Create a transfer of the Replication type that will use the previously created source and target endpoints.

  4. Activate the transfer and wait for its status to change to Replicating.

Test your transfer

  1. In a separate terminal, run kafkacat in consumer mode:

    kafkacat \
    -C \
    -b <broker_host_1_FQDN>:9091,...,<broker_host_N_FQDN>:9091 \
    -t cdc.db1.measurements \
    -X security.protocol=SASL_SSL \
    -X sasl.mechanisms=SCRAM-SHA-512 \
    -X sasl.username=kafka-user \
    -X sasl.password=<password> \
    -X ssl.ca.location=/usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt \
    -Z \
    -K:

    You can get the FQDNs of broker hosts with the list of hosts in the Managed Service for Apache Kafka® cluster.

  2. Connect to the source cluster and populate the measurements table with data:

    INSERT INTO db1.measurements VALUES
    ('iv7b74th678t********', '2022-06-08 17:45:00', 53.70987913, 36.62549834, 378.0, 20.5, 5.3, 20, NULL),
    ('iv9a94th678t********', '2022-06-07 15:00:10', 55.70985913, 37.62141918,  417.0, 15.7, 10.3, 17, NULL);

  3. Make sure the terminal running kafkacat displays the data format schema of the db1.measurements table and information about the added rows.

    Message snippet example
    {
    "payload": {
        "device_id": "iv7b74th678t********"
    },
    "schema": {
        "fields": [
            {
                "field": "device_id",
                "optional": false,
                "type": "string"
            }
        ],
        "name": "cdc.db1.measurements.Key",
        "optional": false,
        "type": "struct"
    }
}: {
    "payload": {
        "after": {
            "altitude": 378,
            "battery_voltage": 5.3,
            "cabin_temperature": 20,
            "datetime": "2020-06-08T17:45:00Z",
            "device_id": "iv7b74th678t********",
            "fuel_level": null,
            "latitude": 53.70987913,
            "longitude": 36.62549834,
            "speed": 20.5
        },
        "before": null,
        "op": "c",
        "source": {
            "connector": "mysql",
            "db": "db1",
            "file": "mysql-log.000016",
            "gtid": "1e46a80b-2e96-11ed-adf7-d00d183780**:*-*****",
            "name": "cdc",
            "pos": 1547357,
            "query": null,
            "row": 0,
            "server_id": 0,
            "snapshot": "false",
            "table": "measurements",
            "thread": null,
            "ts_ms": 1662632515000,
            "version": "1.1.2.Final"
        },
        "transaction": null,
        "ts_ms": 1662632515000
    },
    "schema": {
        "fields": [
            {
                "field": "before",
                "fields": [
                    {
                        "field": "device_id",
                        "optional": false,
                        "type": "string"
                    },
                    ...
                ],
                "name": "cdc.db1.measurements.Value",
                "optional": true,
                "type": "struct"
            },
            {
                "field": "after",
                "fields": [
                    ...
                ],
                "name": "cdc.db1.measurements.Value",
                "optional": true,
                "type": "struct"
            },
            {
                "field": "source",
                "fields": [
                    {
                        "field": "version",
                        "optional": false,
                        "type": "string"
                    },
                    ...
                ],
                "name": "io.debezium.connector.mysql.Source",
                "optional": false,
                "type": "struct"
            },
            {
                "field": "op",
                "optional": false,
                "type": "string"
            },
            ...
        ],
        "name": "cdc.db1.measurements.Envelope",
        "optional": false,
        "type": "struct"
    }
}

Features of transferring data with Data Transfer

  • When transferring data from MySQL® to Apache Kafka®, certain data types undergo conversion:

    • tinyint(1) becomes boolean.
    • real becomes double.
    • bigint unsigned becomes int64.

  • In the payload.source metadata section, server_id and thread remain blank.

Delete the resources you created

Some resources are not free of charge. Delete the resources you no longer need to avoid paying for them:

  1. Deactivate and delete the transfer.

  2. Delete the endpoints.

  3. Delete the clusters:

  4. If you used static public IP addresses to access the cluster hosts, release and delete them.

Previous
Transferring data from Yandex Managed Service for Apache Kafka® using Yandex Data Transfer
Next
Transferring data to Yandex Managed Service for Apache Kafka® using Debezium