How to sync data from MySQL to Google BigQuery using Debezium and Kafka Connect

Why use Debezium and Kafka Connect?

Debezium and Kafka Connect are open-source platforms that provide a powerful solution for streaming data changes in real-time between systems. That real-time interaction allows you to keep your data in sync and accessible for various use cases such as real-time analytics, data warehousing, and data pipeline integrations.

Technology used

Before we go into the details of setting up Debezium and Kafka Connect to sync data from MySQL to BigQuery, it is important to understand the technologies that you will be using and how they are connected.

Change data capture

Change data capture (CDC) is a technique for capturing and recording all the changes made to a database over time. This allows for real-time data replication, making it easy to keep multiple systems in sync.

CDC does this by detecting row-level changes in database source tables, which are characterized as “Insert,” “Update,” and “Delete” events. CDC then notifies other systems or services that rely on the same data.​

Apache Kafka

Apache Kafka is a distributed streaming platform that is used for building real-time data pipelines and streaming applications. It allows for the storage and processing of streams of records in a fault-tolerant way.

Kafka Connect

Kafka Connect is a framework for connecting Kafka with external systems such as databases, key-value stores, search indexes, and file systems, using so-called connectors.​

Kafka connectors are ready-to-use components that can help you to import data from external systems into Kafka topics and export data from Kafka topics into external systems. You can use existing connector implementations for common data sources and syncs or implement our own connectors.​

Debezium

Debezium is an open-source platform that allows you to easily stream changes from your MySQL database to other systems using CDC. It works by reading MySQL binlog to capture data changes in a transactional manner, so you can be sure that you’re always working with the most up-to-date data.

By using Debezium, you can capture the changes made to the MySQL database and stream them to Kafka. Data on the changes can then be consumed by Kafka Connect to load the data into BigQuery.

BigQuery setup

1. 1.Creating a BigQuery project and dataset:
   1. 1.In the Google Cloud Console, navigate to the BigQuery page and create a new project (Creating and managing projects | Resource Manager Documentation | Google Cloud). We will name it “mysql-bigquery” for this tutorial.
   2. 2.Within the project, create a new dataset (Creating datasets | BigQuery | Google Cloud). We will name it “debezium” for this tutorial.
   3. 3.Note that Debezium will automatically create tables in the dataset that match the structure of the MySQL tables being monitored.
2. 2.Creating a GCP service account with BigQuery editor role:
   1. 1.In the Google Cloud Console, navigate to the IAM & Admin page and create a new service account (Creating and managing service accounts | IAM Documentation | Google Cloud).
   2. 2.Give the service account a name and description, then select the “BigQuery Data Editor” role.
3. 3.Generating and downloading a key for the service account:
   1. 1.In the Google Cloud Console, navigate to the IAM & Admin page, find the service account, and click on the three dots on the right, then select "create key" (Create and manage service account keys | IAM Documentation | Google Cloud).
   2. 2.Select JSON as the key type and download the key file.
   3. 3.Store the key file securely and use it to authenticate the connector in Kafka Connect when accessing the BigQuery dataset.

Tutorial

To start syncing data from MySQL to BigQuery we will need following components:

• Apache Zookeeper.
• Apache Kafka.
• Kafka Connect/Debezium service with MySQL connector and Google BigQuery connector plugins.
• MySQL database.

Start required services

1. 1.

   Let’s start with creating a new directory. Open Terminal and run:

   $ mkdir mysql-to-bigquery $ cd mysql-to-bigquery
2. 2.

   Create a plugins directory

   $ mkdir plugins
3. 3.

   Download Debezium mysql plugin:

   $ wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/2.1.1.Final/debezium-connector-mysql-2.1.1.Final-plugin.tar.gz -O mysql-plugin.tar.gz $ tar -xzf mysql-plugin.tar.gz -C plugins
4. 4.

   Download BigQuery plugin and put the contents into your plugins directory (in this tutorial we are using version v2.4.3). Now your plugins directory should look like this:

   $ ls plugins debezium-connector-mysql wepay-kafka-connect-bigquery-2.4.3
5. 5.

   Create a new file (“docker-compose.yml”) with these configurations:

   version: '2' services: zookeeper: container_name: zookeeper image: quay.io/debezium/zookeeper:2.1 ports: - 2181:2181 - 2888:2888 - 3888:3888 kafka: container_name: kafka image: quay.io/debezium/kafka:2.1 ports: - 9092:9092 links: - zookeeper environment: - ZOOKEEPER_CONNECT=zookeeper:2181 mysql: container_name: mysql image: quay.io/debezium/example-mysql:2.1 ports: - 3306:3306 environment: - MYSQL_ROOT_PASSWORD=debezium - MYSQL_USER=mysqluser - MYSQL_PASSWORD=mysqlpw connect: container_name: connect image: quay.io/debezium/connect-base:2.1 volumes: - ./plugins:/kafka/connect ports: - 8083:8083 links: - kafka - mysql environment: - BOOTSTRAP_SERVERS=kafka:9092 - GROUP_ID=1 - CONFIG_STORAGE_TOPIC=my_connect_configs - OFFSET_STORAGE_TOPIC=my_connect_offsets - STATUS_STORAGE_TOPIC=my_connect_statuses
6. 6.

   Let’s start the services:

   $ docker-compose up
7. 7.

   You should see an output similar to the following:

   ... 2023-01-16 15:48:33,939 INFO || Kafka version: 3.0.0 [org.apache.kafka.common.utils.AppInfoParser] ... 2023-01-16 15:48:34,485 INFO || [Worker clientId=connect-1, groupId=1] Starting connectors and tasks using config offset -1 [org.apache.kafka.connect.runtime.distributed.DistributedHerder] 2023-01-16 15:48:34,485 INFO || [Worker clientId=connect-1, groupId=1] Finished starting connectors and tasks [org.apache.kafka.connect.runtime.distributed.DistributedHerder]
8. 8.

   Check if Debezium is running with Kafka Connect API. An empty array in response shows that there are no connectors currently registered with Kafka Connect.

   $ curl -i -X GET -H "Accept:application/json" localhost:8083/connectors []
9. 9.

   We also have MySQL running with an example database inventory. You can check what tables are there by running:

   $ docker exec -it mysql mysql -uroot -pdebezium -D inventory -e "SHOW TABLES;" +——————————-+ | Tables_in_inventory | +——————————-+ | addresses | | customers | | geom | | orders | | products | | products_on_hand | +——————————-+

Let’s check what’s inside the customers table:

Configure Debezium to start syncing MySQL to Kafka

Now let’s configure Debezium to start syncing the inventory database with Kafka.

1. 1.

   Create a new file (“register-mysql.json”) with these configurations (You can find information about these configuration properties in the Debezium documentation):

   { "name": "inventory-connector-mysql", "config": { "connector.class": "io.debezium.connector.mysql.MySqlConnector", "tasks.max": "1", "database.hostname": "mysql", "database.port": "3306", "database.user": "root", "database.password": "debezium", "database.server.id": "184054", "topic.prefix": "debezium", "database.include.list": "inventory", "schema.history.internal.kafka.bootstrap.servers": "kafka:9092", "schema.history.internal.kafka.topic": "schemahistory.inventory" } }
2. 2.

   Register a MySQL connector:

   $ curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://localhost:8083/connectors/ -d @register-mysql.json
3. 3.

   Verify that “inventory-connector” is included in the list of connectors:

   $ curl -H "Accept:application/json" localhost:8083/connectors/ ["inventory-connector-mysql"]
4. 4.

   You can now see database contents in Kafka. To see topics, run:

   $ docker exec -it kafka bash bin/kafka-topics.sh —list —bootstrap-server kafka:9092 ... debezium.inventory.addresses debezium.inventory.customers ...

Let’s check debezium.inventory.addresses:

For more information on Debezium events, see this Debezium documentation.

Configure Debezium to start syncing data to Google BigQuery

Before you start configuring the BigQuery connector, move the Google BigQuery service account key file (details in previous section) to your working directory and name it “bigquery-keyfile.json”.

1. 1.

   Once you have the key file, copy it to the Connect container:

   $ docker cp bigquery-keyfile.json connect:/bigquery-keyfile.json
2. 2.

   Now create a file register-bigquery.json with these configurations (You can find information about these configuration properties in the official documentation):

   { "name": "inventory-connector-bigquery", "config": { "connector.class": "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector", "tasks.max": "1", "consumer.auto.offset.reset": "earliest", "topics.regex": "debezium.inventory.*", "sanitizeTopics": "true", "autoCreateTables": "true", "keyfile": "/bigquery-keyfile.json", "schemaRetriever": "com.wepay.kafka.connect.bigquery.retrieve.IdentitySchemaRetriever", "project": "mysql-bigquery", "defaultDataset": "debezium", "allBQFieldsNullable": true, "allowNewBigQueryFields": true, "transforms": "regexTopicRename,extractAfterData", "transforms.regexTopicRename.type": "org.apache.kafka.connect.transforms.RegexRouter", "transforms.regexTopicRename.regex": "debezium.inventory.(.*)", "transforms.regexTopicRename.replacement": "$1", "transforms.extractAfterData.type": "io.debezium.transforms.ExtractNewRecordState" } }
3. 3.

   To register the BigQuery connector, run:

   $ curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://localhost:8083/connectors/ -d @register-bigquery.json
4. 4.

   Verify that “inventory-connector” is included in the list of connectors:

   $ curl -H "Accept:application/json" localhost:8083/connectors/ ["inventory-connector-mysql","inventory-connector-bigquery"]

In your BigQuery dataset, you will now be able to see tables matching those in MySQL.

Now, select data from your customers table. Emails used are for example purposes only and do not correspond to real individuals.

You can create a new entry in MySQL customers table:

You will see that a new entry has automatically synced to BigQuery.

Conclusion

You should now have a clear understanding of the benefits of syncing data from MySQL to BigQuery using Debezium and Kafka Connect. With the detailed tutorial found in this article, you will be able to set up and configure Debezium and Kafka Connect yourself.

As a reminder, it’s important to test and monitor the pipeline to ensure that data is being synced as expected and to troubleshoot any issues that may arise.

For more information on Debezium and Kafka Connect, you can refer to the following resources:

• Debezium documentation
• Kafka Connect documentation
• BigQuery connector documentation