[Kafka-connect] Streaming the data of MySQL throughs Kafka-connect and Debezium plugin.

Contents

1) Install and configurate mysql to use to the kafka-connect Lab.

Run docker MySQL:

docker run -p 3306:3306 --name mastermysql -e MYSQL_ROOT_PASSWORD=123456 -d mysql:5.7

docker exec -it mastermysql bash
apt update -y
apt install vim -y
cd /etc/mysql/

giờ sửa file: mysql.cnf

vi mysql.cnf

[mysqld]
server-id = 42
log_bin = mysql-bin
binlog_format = row
expire_logs_days = 10

xong rồi thoát ra restart docker

docker restart mastermysql
docker exec -it mastermysql bash

mysqladmin variables -uroot -p|grep log_bin

In order for Debezium MySQL to track all the changes in bin-log, it needs a user with SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT permissions. Grant them to the user you intend to use for Debezium.

giờ chúng ta tạo 1 User để cho kafka-connect.
ở dưới là user: debezium và pass là: 123456

>>>Mysql 5.7
mysql -uroot -p
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE,
REPLICATION CLIENT ON *.* TO 'debezium' IDENTIFIED BY '123456';

>>> Mysql 8
docker run -p 3306:3306 --name mastermysql -e MYSQL_ROOT_PASSWORD=123456 -d mysql:8.0

CREATE USER 'debezium'@'%' IDENTIFIED BY '123456';
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'debezium'@'%';

Xong rồi mình tạo database và table như hình bên dươi.

2) Config kafka-connect.

2.1) Repairing

Mình đã hướng dẫn càc bạn cài các components sau:

Kafka:
https://nimtechnology.com/2022/02/13/kafka-install-kafka-and-zookeeper-cluster-on-kubernetes/
Hoặc starting kafka và zookeeper bằng command lại để lab.
https://nimtechnology.com/2022/01/30/kafka-zookeeper-starting-kafka-and-zookeeper/

Kafka-connect:
https://nimtechnology.com/2022/02/09/kafka-connect-install-kafka-connect-on-kubernetes-through-helm-chart/

Schema registry:
https://nimtechnology.com/2022/02/13/kafka-installing-schema-registry-to-use-for-the-kafka-and-kafka-connect-model/

Lense nữa:
https://nimtechnology.com/2022/02/09/kafka-connect-install-lenses-to-manage-kafka/
cái này thì cần lincense.
Bạn có thể dụng câu lệnh curl trực tiếp vào kafka connect nhé. Command sao mình chỉ sau he.

Reference Links:
https://dev-yeye.tistory.com/35?category=904307

Bạn chuẩn bị đầu đủ các phần trên nhé

2.2) Action config

Giờ login vào lenses.
tạo connector

Vậy CDC là gì?

Đầu tiên cdc là Change Data Capture (CDC)
Ở bước tạo mysql mình có tạo user debezium để nó thể đọc được log_bin, nghĩa là các thay đổi trong database.
nếu bạn làm mysql master-slave rồi thì thấy cách cdc này nó cũng như thế.

reference links for master:
https://developer.confluent.io/learn-kafka/data-pipelines/kafka-data-ingestion-with-cdc/

Nếu lenses của các bạn ko có icon giống mình thì đọc lại bài kafka-connect và build lại image nhé.
Links add plugin for kafka connect

giờ chúng ta cần copy data và mình cung cấp và paste vào đó

Config sau là dành cho mysql 5.7

connector.class=io.debezium.connector.mysql.MySqlConnector
database.user=debezium
database.server.id=223344
database.history.kafka.topic=dbhistory.voting_contra.v2
database.history.kafka.bootstrap.servers=kafka:9092
database.server.name=connect_wordpress_note
database.port=3306
include.schema.changes=false
table.whitelist=wordpress_note.students
database.hostname=192.168.101.27
database.password=***********
name=cdc-wordpress-note-connector
database.whitelist=wordpress_note

Config sau là dành cho mysql 8.0

connector.class=io.debezium.connector.mysql.MySqlConnector
database.allowPublicKeyRetrieval=true
database.user=debezium
database.server.id=223344
database.history.kafka.topic=dbhistory.school.v1
database.history.kafka.bootstrap.servers=kafka:9092
database.server.name=connect_school
database.port=3306
include.schema.changes=false
table.whitelist=school.students
database.hostname=192.168.101.27
database.password=***********
name=cdc-school-connector
database.whitelist=school

Note config with mysql 8.x
Bạn sẽ gặp lỗi này “Public Key Retrieval Is Not Allowed” hay đọc bài viết bên dứoi
https://rmoff.net/2019/10/23/debezium-mysql-v8-public-key-retrieval-is-not-allowed/

curl -i -X PUT -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/source-debezium-mysql-00/config \
    -d '{
          "connector.class": "io.debezium.connector.mysql.MySqlConnector",
          "database.hostname": "mysql",
          "database.port": "3306",
          "database.user": "debezium",
          "database.password": "dbz",
          "database.server.id": "42",
          "database.allowPublicKeyRetrieval":"true", <<<==== add this line to FIX Error
          "database.server.name": "asgard",
          "table.whitelist": "demo.customers",
          "database.history.kafka.bootstrap.servers": "kafka:29092",
          "database.history.kafka.topic": "asgard.dbhistory.demo" ,
          "include.schema.changes": "true"
    }'

Explain the configurations

name: đơn giản là name của connector đó.

connector.class
cái này chỗ khai báo các plugin của chúng ta.

database.port: Port của DB
database.hostname: IP hoặc hostname của DB
database.user: username mà debezium để đọc log_bin
database.password: password của user trên

database.server.id: số này bạn cứ để số nào cũng được nhưng duy nhất nhé (unique) trên cluster kafka, Nó chỉ có ý nghĩa trên kafka thôi

database.whitelist: đây là cái database bạn muốn stream data
table.whitelist: đây là table mà bạn muốn stream data

database.server.name
table.whitelist
với cặp ở trên mình thấy nó có liên quan đến Schema registry

database.history.kafka.topic
chỗ này mình thấy tạo ra 1 topic như sau:

Nhìn sơ giống như lưu lại các bước drop, create database, table, ….

database.history.kafka.bootstrap.servers: cái này để bạn để IP và port của kafka

Các bạn hỏi ủa rồi nó có stream data trong table đó vào topic hem????

Mình thấy có topic này: **<database.server.name>.**<**table.whitelist>**

Bạn sẽ thấy là trong topic này đã có 3 record nghĩ là trong table của cũng đang có 3 row.

giờ mình sẽ thêm 1 row data nữa.

Mình xin phép add thêm ảnh của case khác để các bạn tham khảo:

2.3) Add a new column inside a table that is streaming.

Giờ cũng table đó mình add thêm 1 column

thì mình thấy là e.history.kafka.topic với topic tên là dbhistory.voting_contra.v2 đã tăng 1 record

Giờ thêm data vào table;

2.4) Create a new table

Vậy giờ chúng ta edit config conector như thế nào?

Bạn thấy mình sửa như thế này.

Khi mình edit và save lại thì thấy có lỗi.

You may attempt to recover it by reconfiguring the connector to SCHEMA_ONLY_RECOVERY

Mình đã dò google:
https://gitter.im/debezium/dev/archives/2019/01/26

Jiri Pechanec @jpechane Jan 26 2019 13:18 UTC
@rajan-g Hi, please look at https://debezium.io/docs/connectors/mysql/ and search for schema_onyl_recovery string

connector.class=io.debezium.connector.mysql.MySqlConnector
database.user=debezium
database.server.id=223344
database.history.kafka.topic=dbhistory.voting_contra.v3
database.history.kafka.bootstrap.servers=kafka:9092
database.server.name=connect_wordpress_note
database.port=3306
include.schema.changes=false
table.whitelist=wordpress_note.students,wordpress_note.company
key.converter.schemas.enable=true
database.hostname=192.168.101.27
database.password=***********
name=cdc-wordpress-note-connector
value.converter.schemas.enable=true
database.whitelist=wordpress_note
snapshot.mode=schema_only_recovery

Note nho nhỏ:
Nếu ở lần đầu tiên create connector mà bạn chơi:
snapshot.mode=schema_only_recovery
thì sẽ bị báo lỗi

Could not find existing binlog information while attempting schema only recovery snapshot

file đã nhiều hơn. mà mình ko thấy lỗi nữa

chưa thấy tạo topic data.

Giờ thì topic data của chúng ta muốn đã được tạo.

2.5) Adding the old tables but before that it wasn’t added into connector.

Mình gặp trường hợp trong database của mình có 2 table:

Tiếp đên mình tạo connector với table company và với config như bên dưới

connector.class=io.debezium.connector.mysql.MySqlConnector
database.user=debezium
database.server.id=11
database.history.kafka.topic=_dbhistory.nimtechnology.v1
database.history.kafka.bootstrap.servers=kafka:9092
database.server.name=connect_nimtechnology
database.port=3306
table.whitelist=wordpress_note.company
database.hostname=192.168.101.27
database.password=123456
name=source.mysqldebezium.nimtechnology.v1
database.whitelist=wordpress_note
include.schema.changes=false

Giờ mình stream (CDC) thêm table students

connector.class=io.debezium.connector.mysql.MySqlConnector
database.user=debezium
database.server.id=11
database.history.kafka.topic=_dbhistory.nimtechnology.v2
database.history.kafka.bootstrap.servers=kafka:9092
database.server.name=connect_nimtechnology
database.port=3306
table.whitelist=wordpress_note.company,wordpress_note.students
database.hostname=192.168.101.27
database.password=123456
name=source.mysqldebezium.nimtechnology.v1
database.whitelist=wordpress_note
include.schema.changes=false
snapshot.mode=schema_only_recovery

Mình đã sửa database.history.kafka.topic và add thêm table students vào line “table.whitelist”

mình thấy tạo them topic history nhưng ko tạo topic Data students

Bạn phài insert data mới vào table students thì kafka connect mới tao topic

Mình show data có trong table.

3) Note.

nó còn 1 phần mình thấy hay được sài đó là:

Event Message Flattening with Single Message Transform
https://rmoff.net/2018/03/24/streaming-data-from-mysql-into-kafka-with-kafka-connect-and-debezium/

mình sẽ tìm hiêu và trình bày sau.

Command check mysql:

mysql -u {user} -p'{pw}' -h {host} -P {port} -D {db} -e "SHOW GRANTS"