Kafka Connect MySQL: A Comprehensive Guide

Introduction

Kafka Connect is a powerful tool for integrating Apache Kafka with external systems, including databases. In this article, we’ll dive into how to set up and use Kafka Connect with a MySQL database.

Setup

To get started with Kafka Connect, you’ll need to have Apache Kafka and the Confluent Platform installed. Then, you’ll need to download the Confluent JDBC Connector for MySQL.

Step 1: Create a MySQL Database

The first step is to create a MySQL database that we can use for our example. To do this, you can use the following SQL commands:

CREATE DATABASE kafka_connect_mysql;

USE kafka_connect_mysql;

CREATE TABLE customers (
  id INT AUTO_INCREMENT PRIMARY KEY,
  first_name VARCHAR(50),
  last_name VARCHAR(50),
  email VARCHAR(100)
);

Step 2: Start the Confluent Platform

Next, start the Confluent Platform using the following command:

confluent start

Step 3: Create a Connector Configuration

Now that the Confluent Platform is running, we can create a connector configuration to connect to our MySQL database. To do this, create a mysql-sink.properties file with the following contents:

name=mysql-sink
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
tasks.max=1
topics=customers
connection.url=jdbc:mysql://localhost:3306/kafka_connect_mysql
connection.user=root
connection.password=root
insert.mode=upsert
pk.fields=id
pk.mode=record_value

Step 4: Start the Connector

Now that we have a connector configuration, we can start the connector using the following command:

confluent load mysql-sink -d mysql-sink.properties

Step 5: Produce Data to the customers Topic

To test our setup, we’ll produce some data to the customers topic. To do this, we’ll use the following command:

echo '{"id": 1, "first_name": "John", "last_name": "Doe", "email": "john.doe@example.com"}' | confluent produce customers

Step 6: Verify the Data in MySQL

Finally, we can verify that the data has been correctly inserted into the MySQL database by using the following SQL command:

SELECT * FROM customers;

Code Example

Here’s the complete code example for setting up and using Kafka Connect with a MySQL database:

# Create the MySQL database
CREATE DATABASE kafka_connect_mysql;

USE kafka_connect_mysql;

CREATE TABLE customers (
  id INT AUTO_INCREMENT PRIMARY KEY,
  first_name VARCHAR(50),
  last_name VARCHAR(50),
  email VARCHAR(100)
);

# Start the Confluent Platform
confluent start

# Create the connector configuration
echo '
name=mysql-sink
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
tasks.max=1
topics=customers
connection.url=jdbc:mysql://localhost:3306/kafka_connect_mysql
connection.user=root
connection.password=root
insert.mode=upsert
pk.fields=id
pk.mode=record_value
' > mysql-sink.properties