Building an Enterprise CDC Solution

In this step, we start two different database engines and enable a CDC process. As a result, you have two Kafka topics with Debezium events that you can consume.

The repository contains a docker-compose to run several services in your local environment:

• Two database engines: MySQL and PostgreSQL
• One Kafka broker (and its zookeeper)
• Two Kafka connect services: one to run CDC Debezium tasks and another to send the events to Snowflake

In a terminal run:

cd services
docker-compose up

It can take several minutes to download and start the services. Keep this terminal open to be able to see the log of services. When the tutorial is finished, you can stop all using Ctrl+C.

There are two SQL initialization scripts:

• database/sql/00_mysql_init.sql: create table users
• database/sql/00_postgres_init.sql: create table product

To apply these SQL scripts it in dockerized services, and populate data, run the following lines in a terminal:

cd database
# create tables
./init_db.sh
# Populate data
./mysql_crud.sh
./postgres_crud.sh

In the output, you can see several CRUD operations in the tables, and the last state after operations. You can close this terminal.

The docker service cdc_connect has the necessary dependencies to run Debezium over MySQL and Postgres. The configuration is available in:

• debezium/connect/debezium-mysql-inventory-connector.json
• debezium/connect/debezium-postgres-inventory-connector.json

Open a terminal and init the capture of the tables:

cd debezium
./init_cdc.sh

In the docker-compose terminal, you can see how the connectors start. When the log stabilizes, you can check the status of the Debezium connectors in the previous terminal using:

# I assume you are in the debezium folder
./status_cdc.sh

You can close this terminal.

You can test if the capture is working with this strategy:

• Open a terminal with live consumer events
• Do CRUD operations in the database

First for MySQL, open a terminal and run:

cd services
docker-compose exec kafka /kafka/bin/kafka-console-consumer.sh 
    --bootstrap-server kafka:9092 --from-beginning 
    --topic mysqldb.inventory.users 
    --property print.key=true --property print.timestamp=true

The terminal will populate every new event pushed from Debezium to Kafka, sending every insert/update/delete done in inventory.users in MySQL

Open now a terminal in PostgreSQL and do the same:

cd services
docker-compose exec kafka /kafka/bin/kafka-console-consumer.sh 
    --bootstrap-server kafka:9092 --from-beginning 
    --topic postgresdb.inventory.product 
    --property print.key=true --property print.timestamp=true

To generate new events, open a terminal and run:

cd database
./mysql_crud.sh
./postgres_crud.sh

You should see new data change events in the consumer terminals.

In this step, you send the Kafka events to Snowflake and generate a replica of the source tables.

The docker service sink_connect has the necessary dependencies to run the Snowflake Sink connector to push new Kafka events into the Snowflake table. The configuration is available in snowflake/connect/snowflake-sink-connector.json and you need to update:

• The Snowflake URL with yours in field snowflake.url.name
• The authentication fields if you generate your key-pair in the previous step: snowflake.private.key and snowflake.private.key.passphrase

Open a terminal and init the upload of the Kafka topics:

cd snowflake
./init_sink.sh

In the docker-compose terminal, you can see how the connector starts. When the log stabilizes, you can check the status of the Snowflake connector in the previous terminal using:

# From the snowflake folder
./status_sink.sh

When the sink connector uploads the events from the Kafka topics, it creates these tables:

• CDC_MYSQL_INVENTORY_USERS
• CDC_POSTGRESDB_INVENTORY_PRODUCT

The upload to Snowflake will be done in batches, so it may take some time until the data is available in Snowflake (in the order of 30-60 seconds).

From your Snowflake Worksheet, validate that your events are populated in the new tables:

USE ROLE ACCOUNTADMIN;
USE SCHEMA HOWTO_DB.PUBLIC;
SELECT * FROM CDC_MYSQL_INVENTORY_USERS;
SELECT * FROM CDC_POSTGRESDB_INVENTORY_PRODUCT;

Adding new changes in your dockerized databases produces new rows in your tables.

1. In the Snowflake Worksheet:

   SELECT 'Events MySQL', COUNT(1) FROM CDC_MYSQL_INVENTORY_USERS
   UNION ALL
   SELECT 'Events PostgreSQL', COUNT(1) FROM CDC_POSTGRESDB_INVENTORY_PRODUCT;

2. From a terminal, apply changes in your databases:

   cd database
   ./mysql_crud.sh
   ./postgres_crud.sh

3. Wait until the events are sent to Snowflake (you can see the log in docker-compose terminal)
4. Repeat the query in the Snowflake Worksheet

In the repository there are two scripts with the SQL logic to generate the replica of the source tables:

• snowflake/sql/01-cdc-to-replica-mysql.sql
• snowflake/sql/01-cdc-to-replica-postgres.sql

From your Snowflake Worksheet, execute these two scripts. As a result, you have two views with the same structure of the source databases:

• MYSQL_INVENTORY_USERS
• POSTGRESDB_INVENTORY_PRODUCT

These scripts follow the same logic, creating a scheduled task that processes the new events that arrive and updates the replica table.

Note: one part of these SQL scripts (the MERGE sentence) depends on  the source database engine. The Debezium events have the metadata about the source engine and are used to know which is the last event for an entity. Take into account if you replicate this logic in your production systems.

The end-to-end is running now. You can check the data available in your local databases and validate it against the Snowflake view:

1. In a terminal, get the actual state of MySQL users table:

   cd services
   echo "SELECT * FROM users ORDER BY id" | docker-compose 
      exec -T mysql 
      bash -c 'mysql -u $MYSQL_USER -p$MYSQL_PASSWORD inventory'

2. Go to the Snowflake Worksheet and validate the result with:

   USE ROLE ACCOUNTADMIN;
   USE SCHEMA HOWTO_DB.PUBLIC;
   SELECT * FROM MYSQL_INVENTORY_USERS;

3. In a terminal, get the actual state of the PostgreSQL product table:

   # I assume you are in the services folder
   echo "SELECT * FROM product ORDER BY id" |  docker-compose 
      exec -T postgres 
      env PGOPTIONS="--search_path=inventory" 
      bash -c 'psql -U $POSTGRES_USER postgres'

4. And validate in the Snowflake Worksheet.

   USE ROLE ACCOUNTADMIN;
   USE SCHEMA HOWTO_DB.PUBLIC;
   SELECT * FROM POSTGRESDB_INVENTORY_PRODUCT;

5. Generate new insert-delete-update operations from a terminal:

   cd database
   ./mysql_crud.sh
   ./postgres_crud.sh

6. Wait until the events are sent to Snowflake (review docker-compose terminal log).
7. Wait until the scheduled task is triggered in Snowflake:

   USE ROLE ACCOUNTADMIN;
   select name, state, error_code, error_message,scheduled_time, next_scheduled_time
      from table(HOWTO_DB.information_schema.task_history())
      order by scheduled_time desc;

8. Validate again the content of the tables in Snowflake

Debezium provides an easy way to capture changes from databases and populate change events in a Kafka service, that you can consume in several ways.

To populate these changes to another database you can use the simplest way, but this is not always the best option in the context of the new generation of data warehouses, and probably you need to take into account another kind of strategy close to this service for better performance and reduce the cost of use.

First of all, study the possibilities, and after testing it with a POC similar to this howto, including some aspects like performance and cost review, proceed to do the next steps (security, naming, automatization, data quality, failover, …).