Change Data Captures CDC from MySQL Database to Kafka with Kafka Connect and Debezium

In this post, look at CDC from MySQL to Kafka topics using Debezium MySQL connector.

Ramu kakarla

Jun. 08, 20 · Tutorial

Likes (4)

Comment

Save

12.5K Views

Introduction

Debezium is an open-source project developed by Red Hat which aims to simplify this process by allowing you to extract changes from various database systems (e.g. MySQL, PostgreSQL, MongoDB) and push them to Kafka

Architecture

Debezium Connectors

Debezium has a library of connectors that capture changes from a variety of databases and produce events with very similar structures, making it easier for the applications to consume and respond to the events regardless of where the changes originated. Debezium currently have the following connectors

MySQL Connector
MongoDB Connector
PostgreSQL Connector
SQL Server Connector
Oracle Connector (Incubating)
Db2(Incubating)
Cassandra(Incubating)

In this article, we are going to capture the change events from MySQL database to the Kafka Topics.

The Debezium MySQL connector tracks the structure of the tables, performs snapshots, transforms bin log events into Debezium change events and records where those events are recorded in Kafka.

Prerequisites:

Kafka installed on OpenShift
Kafka connect installed on OpenShift
MySQL Server installed and setup

I have Installed one node Kafka cluster with Kafka connect

    Java
   
x

[kkakarla@kkakarla amq-streams-1.4.0-ocp-install-examples]$ oc get pods
NAME                                          READY   STATUS      RESTARTS   AGE
my-cluster-entity-operator-759bfbc89d-s68wx   3/3     Running     0          9m
my-cluster-kafka-0                            2/2     Running     2          9m55s
my-cluster-zookeeper-0                        2/2     Running     0          11m
my-cluster-zookeeper-1                        2/2     Running     0          11m
my-cluster-zookeeper-2                        2/2     Running     0          11m
my-connect-cluster-connect-1-build            0/1     Completed   0          6m4s
my-connect-cluster-connect-1-deploy           0/1     Completed   0          5m27s
my-connect-cluster-connect-1-ltphz            1/1     Running     0          5m25s
strimzi-cluster-operator-64d9cf9bc7-wsf4q     1/1     Running     0          15m

Installing the MySQL Connector

Installing the Debezium MySQL connector is a simple process; just download the JAR, extract it to the Kafka Connect environment, and ensure the plugin’s parent directory is specified in your Kafka Connect environment.

Download the Debezium MySQL connector.
Extract the files into your Kafka Connect environment.
Add the plugin’s parent directory to your Kafka Connect plugin path:
Create a directory with Kafka Connect plug-ins

    Java
   
xxxxxxxxxx

[kkakarla@kkakarla upstream-plugins]$ tree
.
└── my-plugins
    └── debezium-connector-mysql
        ├── antlr4-runtime-4.7.2.jar
        ├── CHANGELOG.md
        ├── CONTRIBUTE.md
        ├── COPYRIGHT.txt
        ├── debezium-api-1.1.2.Final.jar
        ├── debezium-connector-mysql-1.1.2.Final.jar
        ├── debezium-core-1.1.2.Final.jar
        ├── debezium-ddl-parser-1.1.2.Final.jar
        ├── LICENSE-3rd-PARTIES.txt
        ├── LICENSE.txt
        ├── mysql-binlog-connector-java-0.19.1.jar
        ├── mysql-connector-java-8.0.16.jar
        └── README.md
2 directories, 13 files

The Kafka Connect S2I image takes your binaries (with plug-ins and connectors) and stores them in the /tmp/kafka-plugins/s2i directory. It creates a new Kafka Connect image from this directory, which can then be used with the Kafka Connect deployment. When started using the enhanced image, Kafka Connect loads any third-party plug-ins from the /tmp/kafka-plugins/s2i directory.

Start a new build of the image

    Java
   
xxxxxxxxxx

[kkakarla@kkakarla upstream-plugins]$ oc start-build my-connect-cluster-connect --from-dir ./my-plugins/
Uploading directory "my-plugins" as binary input for the build ...
....................
Uploading finished
build.build.openshift.io/my-connect-cluster-connect-2 started

Once the build has completed, the new image is used automatically by the Kafka Connect deployment

    Java
   
xxxxxxxxxx

[kkakarla@kkakarla upstream-plugins]$ oc get build
NAME                           TYPE     FROM     STATUS     STARTED          DURATION
my-connect-cluster-connect-1   Source            Complete   59 minutes ago   36s
my-connect-cluster-connect-2   Source   Binary   Complete   6 minutes ago    2m7s

Installing MySQL

Install MySQL using command oc new-app --name=mysql debezium/example-mysql:1.0

    Shell
   
x

[kkakarla@kkakarla upstream-plugins]$ oc new-app --name=mysql debezium/example-mysql:1.0
--> Found container image d8322b2 (3 days old) from Docker Hub for "debezium/example-mysql:1.0"
    * An image stream tag will be created as "mysql:1.0" that will track this image
    * This image will be deployed in deployment config "mysql"
    * Ports 3306/tcp, 33060/tcp will be load balanced by service "mysql"
      * Other containers can access this service through the hostname "mysql"
    * This image declares volumes and will default to use non-persistent, host-local storage.
      You can add persistent volumes later by running 'oc set volume dc/mysql --add ...'
    * WARNING: Image "debezium/example-mysql:1.0" runs as the 'root' user which may not be permitted by your cluster administrator
--> Creating resources ...
    imagestream.image.openshift.io "mysql" created
    deploymentconfig.apps.openshift.io "mysql" created
    service "mysql" created
--> Success
    Application is not exposed. You can expose services to the outside world by executing one or more of the commands below:
     'oc expose svc/mysql' 
    Run 'oc status' to view your app.

Specify one of MYSQL_ROOT_PASSWORD, MYSQL_ALLOW_EMPTY_PASSWORD and MYSQL_RANDOM_ROOT_PASSWORD

    Shell
   
xxxxxxxxxx

[kkakarla@kkakarla upstream-plugins]$ oc set env dc/mysql MYSQL_ROOT_PASSWORD=debezium  MYSQL_USER=mysqluser MYSQL_PASSWORD=mysqlpw
deploymentconfig.apps.openshift.io/mysql updated
oc get pods -l app=mysql
NAME            READY   STATUS    RESTARTS   AGE
mysql-2-v2gbg   1/1     Running   0          64s

Connect to MySQL database

    Shell
   
xxxxxxxxxx

oc exec mysql-2-v2gbg -it -- mysql -u mysqluser -p mysqlpw inventory
mysql -u root -p
mysql> show databases;
+--------------------+
| Database           |
+--------------------+
| information_schema |
| inventory          |
| mysql              |
| performance_schema |
| sys                |
+--------------------+
5 rows in set (0.00 sec)
mysql> use inventory
Reading table information for completion of table and column names
You can turn off this feature to get a quicker startup with -A
Database changed
mysql> show tables;
+---------------------+
| Tables_in_inventory |
+---------------------+
| addresses           |
| customers           |
| geom                |
| orders              |
| products            |
| products_on_hand    |
+---------------------+
6 rows in set (0.00 sec)
  
 mysql> select * from customers;
+------+------------+-----------+-----------------------+
| id   | first_name | last_name | email                 |
+------+------------+-----------+-----------------------+
| 1001 | Sally      | Thomas    | sally.thomas@acme.com |
| 1002 | George     | Bailey    | gbailey@foobar.com    |
| 1003 | Edward     | Walker    | ed@walker.com         |
| 1004 | Anne       | Kretchmar | annek@noanswer.org    |
+------+------------+-----------+-----------------------+
4 rows in set (0.00 sec)

Create a MySQL connector instance to monitor the inventory database

    YAML
   
xxxxxxxxxx

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaConnector
metadata:
  name: inventory-connector  
  labels:
    strimzi.io/cluster: my-connect-cluster
spec:
  class: io.debezium.connector.mysql.MySqlConnector
  tasksMax: 1  
  config:  
    database.hostname: mysql  
    database.port: 3306
    database.user: debezium
    database.password: dbz
    database.server.id: 184054  
    database.server.name: dbserver1  
    database.whitelist: inventory  
    database.history.kafka.bootstrap.servers: my-cluster-kafka-bootstrap:9092 
    database.history.kafka.topic: schema-changes.inventory

Inventory-connector.yaml

Verify that inventory-connector was created and has started to monitor the inventory database.

    Java
   
xxxxxxxxxx

oc logs $(oc get pods -o name -l strimzi.io/name=my-connect-cluster-connect)

Capture Data Change Events

Insert

    Java
   
xxxxxxxxxx

INSERT INTO `inventory`.`customers`
(
    `first_name`,
    `last_name`,
    `email`)
VALUES
(
    'Ramu',
    'kakarla',
    'Ramu@gmail.com'
)
  
  "payload":{
      "before":null,
      "after":{
         "id":1005,
         "first_name":"Ramu",
         "last_name":"kakarla",
         "email":"Ramu@gmail.com"
      },
      "source":{
         "version":"1.1.2.Final",
         "connector":"mysql",
         "name":"dbserver1",
         "ts_ms":1591593372000,
         "snapshot":"false",
         "db":"inventory",
         "table":"customers",
         "server_id":223344,
         "gtid":null,
         "file":"mysql-bin.000003",
         "pos":364,
         "row":0,
         "thread":3,
         "query":null
      },
      "op":"c",
      "ts_ms":1591593372420,
      "transaction":null
   }
}

The before object is null while the after object shows the newly inserted value. Notice that the op attribute value is c, meaning it’s a CREATE event.

Update

    Java
   
xxxxxxxxxx

mysql> UPDATE customers SET email='ramu@123@msn.com' WHERE id=1005;
Query OK, 1 row affected (0.02 sec)
Rows matched: 1  Changed: 1  Warnings: 0
  
 "payload":{
      "before":{
         "id":1005,
         "first_name":"Ramu",
         "last_name":"kakarla",
         "email":"Ramu@gmail.com"
      },
      "after":{
         "id":1005,
         "first_name":"Ramu",
         "last_name":"kakarla",
         "email":"ramu@123@msn.com"
      },
      "source":{
         "version":"1.1.2.Final",
         "connector":"mysql",
         "name":"dbserver1",
         "ts_ms":1591594059000,
         "snapshot":"false",
         "db":"inventory",
         "table":"customers",
         "server_id":223344,
         "gtid":null,
         "file":"mysql-bin.000003",
         "pos":673,
         "row":0,
         "thread":8,
         "query":null
      },
      "op":"u",
      "ts_ms":1591594059965,
      "transaction":null
   }
}

The before object shows the row state, while the after object shows the current state of the row. Notice that the op attribute value is u, meaning it’s a UPDATE event.

DELETE

    Java
   
xxxxxxxxxx

mysql> DELETE FROM `inventory`.`customers` WHERE id = 1005;
Query OK, 1 row affected (0.01 sec)
  
   "payload":{
      "before":{
         "id":1005,
         "first_name":"Ramu",
         "last_name":"kakarla",
         "email":"ramu@123@msn.com"
      },
      "after":null,
      "source":{
         "version":"1.1.2.Final",
         "connector":"mysql",
         "name":"dbserver1",
         "ts_ms":1591594605000,
         "snapshot":"false",
         "db":"inventory",
         "table":"customers",
         "server_id":223344,
         "gtid":null,
         "file":"mysql-bin.000003",
         "pos":1018,
         "row":0,
         "thread":8,
         "query":null
      },
      "op":"d",
      "ts_ms":1591594605296,
      "transaction":null
   }
}

The before object state is not null while the after object state is null. Notice that the op attribute value is d, meaning it’s a DELETE event.

Enjoy!

Database MySQL kafka Connector (mathematics) Event Data (computing) Java (programming language)

Opinions expressed by DZone contributors are their own.

Related

Trending