MySQL® change data capture and delivery to Apache Kafka®
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:
- Prepare the source cluster.
- Prepare the target cluster.
- Prepare and activate the transfer.
- Test the transfer.
If you no longer need the resources you created, delete them.
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
- Database:
-
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:
-
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 roleALL_PRIVILEGES
for thedb1
database and grant the global privilegesREPLICATION CLIENT
andREPLICATION SLAVE
. -
Connect to the database
db1
on behalf ofmy-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);
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.
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 theACCESS_ROLE_CONSUMER
andACCESS_ROLE_PRODUCER
roles for the created topics. To include all such topics, putcdc.*
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 theACCESS_ROLE_CONSUMER
andACCESS_ROLE_PRODUCER
roles for topics whose names begin with thecdc
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 Storage in Managed Service for Apache Kafka®.
Prepare 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: Specify the password for
my-user
. - Included tables:
db1.measurements
.
- Connection settings:
- Database type:
-
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: Enter 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 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/YandexInternalRootCA.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 ('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 is displaying thedb1.measurements
table data format schema and information on the inserted 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
-
Transferring data from MySQL® to Apache Kafka® modifies some data types:
tinyint(1)
transfers asboolean
.real
transfers asdouble
.bigint unsigned
transfers asint64
.
-
Under
payload.source
source metadata, omit theserver_id
andthread
parameters.
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 clusters:
-
If static public IP addresses were used for accessing the cluster hosts, release and delete them.