Delivering data from Yandex Managed Service for MySQL to Yandex Managed Service for Apache Kafka®

Written by

Before you begin
Prepare the source cluster
Prepare the target cluster
Prepare and activate the transfer
Test the 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 these resources, delete them.

Before you begin

Create a source Managed Service for MySQL cluster in any suitable configuration with the following settings:
- With the db1 database.
- With a user named my-user.
- With publicly available hosts.
Create a Managed Service for Apache Kafka® target cluster in any applicable configuration with publicly available hosts.
Configure security groups for your clusters so you can connect to them from the internet:
- Instructions for Managed Service for MySQL
- Instructions 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
```

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 user the role ALL_PRIVILEGES for the db1 database and grant the global privileges REPLICATION CLIENT and REPLICATION SLAVE.
Connect to the database db1 on behalf of my-user.

Add test data to the database. As an example, we'll 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
    ('iv9a94th6rztooxh5ur2', '2022-06-05 17:27:00', 55.70329032, 37.65472196,  427.5,    0, 23.5, 17, NULL),
    ('rhibbh3y08qmz3sdbrbu', '2022-06-06 09:49:54', 55.71294467, 37.66542005, 429.13, 55.5, NULL, 18, 32);

Prepare the target cluster

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.

Interfaces Yandex Cloud

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 the ACCESS_ROLE_CONSUMER and ACCESS_ROLE_PRODUCER roles for the topics created. To include all such topics, put cdc.* in the topic name.

If topics are managed using the Kafka Admin API:

Create an administrator-user named kafka-user.
In addition to ACCESS_ROLE_ADMIN, assign the administrative user the ACCESS_ROLE_CONSUMER and ACCESS_ROLE_PRODUCER roles for 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. For more information, see Disk types in Managed Service for Apache Kafka®.

Prepare and activate the transfer

Create an endpoint for the MySQL source with the following settings:
- Data base type: MySQL.
- Endpoint parameters:
  - Connection settings: MDB cluster.
  - MDB cluster: Select the Managed Service for MySQL cluster you created earlier.
  - Database name: db1.
  - Username: my-user.
  - Password: Specify the password for my-user.
  - List of included tables: db1.measurements.
Create an endpoint for the Apache Kafka® target with the following settings:
- DB type: Kafka.
- Endpoint parameters:
  - Connection: Managed Kafka.
  - Managed Kafka:
    - Managed Kafka cluster ID: Select the previously created Managed Service for Apache Kafka® cluster.
    - Authentication: Enter the details of the created kafka-user user.
  - Apache Kafka topic settings: Topic full name.
  - Topic full name: cdc.db1.measurements.
  If you need to track changes in multiple tables, complete the fields as follows:
  - Apache Kafka topic settings: Topic prefix.
  - Topic prefix: Enter the cdc prefix you used to generate topic names.
Create a transfer of type Increment with the previously created source and target endpoints.
Activate the transfer and wait for its status to change to Replicating.

Test the transfer

In a separate terminal, run the kafkacat utility in consumer mode:

kafkacat \
    -C \
    -b <FQDN of broker host 1>:9091,...,<FQDN of broker host N>: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/YandexCA.crt \
    -Z \
    -K:

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

Connect to the source cluster and add data to the measurements table:

INSERT INTO db1.measurements VALUES
    ('iv7b74th678tooxh5ur2', '2022-06-08 17:45:00', 53.70987913, 36.62549834, 378.0, 20.5, 5.3, 20, NULL),
    ('iv9a94th678tooxh5ur2', '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 is displaying the db1.measurements table data format schema and information on the inserted rows.

Example of the message fragment

{
    "payload": {
        "device_id": "iv7b74th678tooxh5ur2"
    },
    "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": "iv7b74th678tooxh5ur2",
            "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-d00d18378058:1-98501",
            "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

Transferring data from MySQL to Apache Kafka® modifies some data types:
- tinyint(1) transfers as boolean.
- real transfers as double.
- bigint unsigned transfers as int64.
Under payload.source source metadata, omit the server_id and thread parameters.

Delete the resources you created

If you no longer need these resources, delete them:

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.