Source
yaml
id: debezium-mysql-realtime-trigger
namespace: company.team
tasks:
- id: if
type: io.kestra.plugin.core.flow.If
condition: "{{ trigger.data.metadata.operation == 'DELETE' }}"
then:
- id: log
type: io.kestra.plugin.core.log.Log
message: Employee {{ trigger.data.first_name }} {{ trigger.data.last_name }} has
been deleted
- id: send_alert
type: io.kestra.plugin.notifications.slack.SlackExecution
url: "{{ secret('SLACK_WEBHOOK') }}"
channel: "#general"
customMessage: Employee {{ trigger.data.first_name }} {{ trigger.data.last_name
}} has been deleted
triggers:
- id: realtime_trigger
type: io.kestra.plugin.debezium.mysql.RealtimeTrigger
snapshotMode: NEVER
includedTables: test.employees
serverId: "12345"
hostname: localhost
port: "13306"
username: mysql_username
password: mysql_password
About this blueprint
Kestra
This flow will:
- Get triggered every time the change data capture event is produced by the MySQL
- If there is a DELETE operation on the table, we would log the employee name that was deleted, and also send out a Slack notificaiton
For creating a MySQL cluster, use the following docker command:
text
docker run -d \
--name mysql \
-v /var/lib/mysql:/var/lib/mysql -p 13306:3306 -e MYSQL_ROOT_PASSWORD=root \
mysql \
mysqld \
--datadir=/var/lib/mysql \
--user=mysql \
--server-id=12345 \
--log-bin=/var/lib/mysql/mysql-bin.log \
--binlog_do_db=test
This docker command puts in the necessary configuration to enable the binlog.
Run the following commands on the MySQL:
text
GRANT REPLICATION SLAVE ON *.* TO 'root'@'%';
# Check if the binlog is enabled
SHOW VARIABLES LIKE 'log_bin';
You can now create the test database and the employees table in MySQL.
text
CREATE DATABASE test;
CREATE TABLE test.employees (id INT, first_name VARCAHR(25), last_name
VARCAHR(25), city VARCAHR(25));
When you insert, update or delete data into test.employees table, the flow will be triggered immediately. Only in case of delete operation, we will log the message and send out a Slack notification.
More Related Blueprints