Enabling change data capture from MySQL to Apache Kafka® with Debezium

Implement a change data capture workflow in Apache Kafka®, a key component of any organization with high data integrity requirements.

Subscribe to RSS

Change Data Capture (CDC) is the process of tracking the changes happening in one or more database table in a source system and propagating them to a target technology. The objective of this tutorial is to create a CDC flow from a source table in a MySQL database to Apache Kafka® via the Kafka Connect Debezium Source connector.

Debezium 2.5

This article describes the configuration for Debezium version 2.5 and later.

Change Data Capture from MySQL to Apache Kafka via Kafka Connect

In order to setup a change data capture process tracking the changes from a MySQL database to Apache Kafka, we can use Kafka Connect, a framework which integrates Apache Kafka with external systems.

There are two types of Kafka Connect connectors we can use to set up a CDC process between MySQL and Apache Kafka:

The JDBC Source connector, a query based approach which periodically pulls the latest changes from MySQL into Kafka.
The Debezium Source connector, which reads from a MySQL binary log (binlog) and streams the changes to Apache Kafka.

While both connectors can move MySQL data into Apache Kafka, the Debezium connector tracks the changes in streaming mode (not in a query based batch mode). This allows us to enhance the extracted data with useful metadata like transaction, IDs and timestamps, and pre-post update details.

The following tutorial guides you through setting up a CDC process using the Debezium connector.

Environment setup

The first step into setting up a CDC process is to create the source MySQL database and the target Apache Kafka cluster with Kafka Connect enabled. We'll use Aiven for MySQL and Aiven for Apache Kafka® to do this.

Create an Aiven account
Install the Aiven Command Line Interface (CLI)
Login into the Aiven CLI using:

Loading code...

Create a Aiven for MySQL service named demo-mysql-source:

Loading code...

Create an Aiven for Apache Kafka service named demo-kafka with Kafka Connect and Schema Registry enabled:

Loading code...

Wait for the services to start with:

Loading code...

Create and populate the MySQL table

Once the services are up and running, we can create the table in MySQL in which we'll track the changes. To create and populate the table:

Get the Aiven for MySQL connection details:

Loading code...

Use the mysql command to connect to the database. Replace the placeholders for <MYSQL_USERNAME>, <MYSQL_PASSWORD>, <MYSQL_HOSTNAME>, <MYSQL_PORT> and <MYSQL_DATABASE_NAME> with the values taken from the command above. Alternatively, you can find the pre-filled mysql command from the Quick Connect button in the Aiven Console service overview.

Loading code...

Note

You can install the mysql command line tool using the MySQL installation guide.

Create a table called users with a numeric id column, a username string and populate with three rows for Francesco, Ana, and Floor

Loading code...

Verify that the above commands are successful by checking the users table with:

Loading code...

The result should be similar to the below:

Loading code...

Create a Kafka Connect Connector

To create the CDC pipeline, we need to define the settings of the Kafka Connect Debezium Source connector. These settings contain:

The source MySQL connection parameters, which allow the connector to fetch the data
The target Apache Kafka connection parameters, which allow the connector to store the versioning of the table structure. This step is optional, but it might be useful to track DDL changes to be able to replicate changes to a target MySQL database.
The target Apache Kafka schema registry connection parameters, which allow the connector to define schemas and store the data in Apache Avro™ format.

Get the connection parameters

As done in the section above, we can use the avn service get command to retrieve the Aiven for MySQL parameters:

Loading code...

We can do the same to retrieve the Apache Kafka Schema Registry parameters needed for the schema integration. You can fetch the connection details with:

Loading code...

Note

Note: the above command uses jq to parse the avn command output and retrieve the connection URI which will be in the format https://<SCHEMA_REGISTRY_USER>:<SCHEMA_REGISTRY_PASSWORD>@<APACHE_KAFKA_HOST>:<SCHEMA_REGISTRY_PORT>. You'll need to parse the information to fill the Kafka Connect connector configuration file in the following section.

To retrieve the Apache Kafka endpoint, which we'll use to track the DDL changes in a topic:

Loading code...

Define the Debezium source connector configuration file

In a file named mysql_source_deb_connector.json add the following JSON configuration to define the Debezium connector configuration:

Loading code...

We need to replace the following values:

<CONNECTOR_NAME> is the name we're giving this connector. For this tutorial, set it to mysql_source_deb_connector.
The MySQL connection details from the previous section: <MYSQL_USERNAME>, <MYSQL_PASSWORD>, <MYSQL_HOSTNAME>, <MYSQL_PORT> and <MYSQL_DATABASE_NAME> and <MYSQL_SSL_MODE>
<UNIQUE_ID> a number, which must be different for each connector running. For this tutorial, set it to 12345.
<KAFKA_TOPIC_PREFIX> will be used as a prefix for all the Kafka topics that this connector sends events to. For this tutorial, set it to example.
<MYSQ_TABLES> is the list of MySQL tables to capture changes from. This parameter is optional. The default is the empty string, which means capture all tables. For this tutorial, set it to defaultdb.users, the one table we have. (Note that there's also a parameter called table.exclude.list for the opposite effect.)
<NR_TASKS> should be set to 1, since the MySQL connector always uses a single task. This is actually the default value, so the tasks.max parameter can also be omitted.
<HISTORY_TOPIC_NAME> is the name of the Apache Kafka topic that contains the history of schema changes. For this tutorial, set it to ddl_history.
The Apache Kafka schema registry details from the previous section: <APACHE_KAFKA_HOST>:<SCHEMA_REGISTRY_PORT> and <SCHEMA_REGISTRY_USER>:<SCHEMA_REGISTRY_PASSWORD>
The Apache Kafka details from the previous section: <APACHE_KAFKA_HOST>:<APACHE_KAFKA_PORT>

See the Debezium MySQL connector documentation for more infomation.

Start the Debezium source connector

Once the parameters are substituted, we can start the Debezium source connector with:

Loading code...

Since we set the name parameter to mysql_source_deb_connector, the above starts a connector with that name.

We can check its status with:

Loading code...

The status should be RUNNING.

Loading code...

Check the data in Apache Kafka with kcat

If the connector is up and running, we can check that the data is flowing from MySQL to Apache Kafka with kcat. Get the kcat parameters and download the certificates needed to connect with:

Loading code...

The above command will store the three certificates (ca.pem, service.crt and service.key) in the current local folder and return the command to execute.

To verify the data in Apache Kafka we can use the following command, substituting the placeholders for <KAFKA_HOST>, <KAFKA_PORT>, <KAFKA_SCHEMA_REGISTRY_USR>, <KAFKA_SCHEMA_REGISTRY_PWD>, <KAFKA_HOST>, and <KAFKA_SCHEMA_REGISTRY_PORT>.

Loading code...

Note

Note: the above command is consuming (-C flag) from the topic example.defaultdb.users (-t flag) in Apache Avro format (-s flag). The topic name is determined by the concatenation of the connector parameter topic.prefix (example) and the table name (defaultdb.users).

After executing the kcat command, the data in Apache Kafka should be similar to the following:

Loading code...

Not only do we get the three values stored in the users table, but we also get additional useful metadata like when the insert/update/deletion happened (in the ts_ms parameter) and the position in the binlog file.

If we now add a new value in MySQL users table with:

Loading code...

We should see it immediately pop up in the Apache Kafka topic:

Loading code...

If we now remove the row containing Carlo with:

Loading code...

We see the corresponding delete ("op": "d") in Apache Kafka:

Loading code...

Altering the schema

Where is the MySQL schema stored? There are two topics:

one called ddl_history (named by the database.history.kafka.topic parameter) which stores the schema for Debezium's inner workings so that Debezium can resume work if it crashes
the other called example (based on the topic.prefix parameter) where the connector stores all the DDLs needed to replicate the state in another MySQL database.

We can fetch the latter topic with kcat:

Loading code...

We get the full list of DDLs needed to replicate the state into a new MySQL database.

Loading code...

Warning

The response also contains database truncation statements which might delete other tables present in the same target database.

If we now add a column to the source MySQL database with:

Loading code...

We get a new entry in the example topic defining the DDL change we just implemented:

Loading code...

Conclusion

The Debezium connector for Apache Kafka tracks, in real time, all the changes happening in one or more MySQL tables into a series of topics by reading the binlog. The messages in Apache Kafka contain the update data and are also enriched with metadata information like the transaction ids and timestamp. Furthermore, the usage of a dedicated topic to track DDL changes enables additional control over the table structures useful to replicate the database structure when needed.

You can dig more with the following resources:

Table of contents

Change Data Capture from MySQL to Apache Kafka via Kafka Connect
Environment setup
Create and populate the MySQL table
Create a Kafka Connect Connector
Get the connection parameters
Define the Debezium source connector configuration file
Start the Debezium source connector
Check the data in Apache Kafka with kcat
Altering the schema
Conclusion