MySQL® change data capture and delivery to Apache Kafka®

Written by

Updated at April 9, 2025

Required paid resources
Getting started
Prepare the source cluster
Prepare the target cluster
Set up and activate the transfer
Test your transfer
- Specifics of data delivery with Data Transfer
Delete the resources you created

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:

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

Required paid resources

The support cost includes:

Managed Service for MySQL® cluster fee: Using computing resources allocated to hosts and disk space (see Managed Service for MySQL® pricing).
Apache Kafka® cluster fee: Using computing resources allocated to hosts (including ZooKeeper hosts) and disk space (see Apache Kafka® pricing).
Fee for using public IP addresses for cluster hosts (see Virtual Private Cloud pricing).
Transfer fee: Use of computing resources and the number of transferred data rows (see Data Transfer pricing).

Getting started

Create a Managed Service for MySQL® source cluster in any suitable configuration with the following settings:
- Database: db1
- Username: my-user
- Hosts: Publicly available
Create a Managed Service for Apache Kafka® target cluster in any suitable configuration with publicly available hosts.
If you are using security groups, configure them to enable connecting to the clusters from the internet:
- Guide for Managed Service for MySQL®
- Guide for Managed Service for Apache Kafka®
Install the kcat (kafkacat) utility and the MySQL command-line tool on the local machine. For example, in Ubuntu 20.04, run:
```
sudo apt update && sudo apt install kafkacat mysql-client --yes
```
Check that you can use it to connect to the Managed Service for Apache Kafka® source cluster over SSL.

Prepare the source cluster

For Data Transfer to get notifications about data updates from a Managed Service for MySQL® cluster, you need to configure external replication in the source cluster. To enable my-user to perform replication, assign the ALL_PRIVILEGES role for the db1 database and issue the REPLICATION CLIENT and REPLICATION SLAVE global privileges to the user.
Connect to the db1 database under my-user.

Add test data to the database. As an example, we will use a simple table with information transmitted by 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);

The settings vary depending on the topic management method used. Data topic names are generated using the following convention: <topic_prefix>.<schema_name>.<table_name>. In this tutorial, the cdc prefix is used as an example.

Yandex Cloud interfaces

Admin API

If topics are managed using standard Yandex Cloud interfaces (management console, YC CLI, Terraform, API):

Create a topic named cdc.db1.measurements.

To track updates to more than one table, create a separate topic for each one.
Create a user named kafka-user with ACCESS_ROLE_CONSUMER and ACCESS_ROLE_PRODUCER roles for the new topics. To include all such topics, put cdc.* in the topic's name.

If topics are managed using the Kafka Admin API:

Create an admin user named kafka-user.
In addition to ACCESS_ROLE_ADMIN, assign the admin user the ACCESS_ROLE_CONSUMER and ACCESS_ROLE_PRODUCER roles for the topics whose names begin with the cdc prefix.

Required topics will be created automatically at the first change event in the tracked tables of a source cluster. This solution can be useful to track changes in multiple tables but requires extra free space in cluster storage. To learn more, see Storage in Managed Service for Apache Kafka®.

Set up and activate the transfer

Create an endpoint for the MySQL® source with the following settings:
- Database type: MySQL.
- Endpoint parameters:
  - Connection settings: Managed Service for MySQL cluster.
    - Managed Service for MySQL cluster: Select the created Managed Service for MySQL® cluster.
    - Database: db1.
    - User: my-user.
    - Password: Enter the my-user password.
    - Included tables: db1.measurements.
Create an endpoint for the Apache Kafka® target 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 created Managed Service for Apache Kafka® cluster.
    * Authentication: Specify the details of the created kafka-user user.
    - 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.
Create a transfer of the Replication type with the created source and target endpoints.
Activate the transfer and wait until its status switches to Replicating.

Test your transfer

In a separate terminal, run the kafkacat utility 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.

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);

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

Example of the message fragment

{
    "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"
    }
}

Specifics of data delivery with Data Transfer

Some data types get modified when transferred from MySQL® to Apache Kafka®:
- tinyint(1) transfers as boolean.
- real transfers as double.
- bigint unsigned transfers as int64.
Under payload.source source metadata, leave server_id and thread blank.

Delete the resources you created

Some resources are not free of charge. To avoid paying for them, delete the resources you no longer need:

Deactivate and delete the transfer.
Delete the endpoints.
Delete the clusters:
- Managed Service for Apache Kafka®
- Managed Service for MySQL®
If static public IP addresses were used for accessing the cluster hosts, release and delete them.

MySQL® change data capture and delivery to Apache Kafka®

Required paid resourcesRequired paid resources

Getting startedGetting started

Prepare the source clusterPrepare the source cluster

Prepare the target clusterPrepare the target cluster

Set up and activate the transferSet up and activate the transfer

Test your transferTest your transfer

Specifics of data delivery with Data TransferSpecifics of data delivery with Data Transfer

Delete the resources you createdDelete the resources you created

Was the article helpful?