Real-time Change Data Capture with Redpanda Connect and MySQL

Your database is constantly changing. Rows get inserted, updated, deleted, all day long. The question is: how does the rest of your stack find out? Polling on a timer is slow and noisy, and bolting "ping the API on every write" into your app code gets ugly fast. There's a better way, and that's what this tutorial will show you.

In the next few minutes, you'll stand up a change data capture (CDC) pipeline that streams row-level changes from MySQL in real time. Every change in your database becomes an event you can react to downstream, no polling required. And here's the nice part: the entire Kafka Connect ecosystem works out of the box with Redpanda, since Redpanda is API-compatible with Apache Kafka^®. So if you've used Kafka Connect before, you already know how to do this. If you haven't, you're about to learn on easy mode.

What is change data capture (CDC)?

‍CDC is the process of recognizing when data has changed in a source system so a downstream process or system can act on that change. Instead of repeatedly asking "anything new yet?", your downstream systems are just notified the moment something happens. New if you're just getting into this stuff? We have a beginner's intro to CDC to get you up to speed.

Why use CDC?

Here are some popular use cases:

• Safe migration from legacy systems
• Division of monolithic data
• Real-time analytics
• Easy integrations
• Cache invalidations
• Outbox patterns

‍Note: Redpanda aims to be fully compatible with Kafka APIs, so all existing connectors should work with Redpanda without any changes.

Tutorial: Real-Time Change Data Capture with Redpanda Connect and MySQL

To run everything smoothly, here's what you need to know before you start.

Prerequisites

• Docker
• Docker Compose
• A Redpanda Connect enterprise license. The mysql_cdc input requires one. You can get a free 30-day trial at redpanda.com/try-enterprise.

Download your license file and save it as redpanda_license.txt. You will copy it into your working directory in the next step.

Quick tour of services

• Redpanda - Stores your change event stream as topics.
• Redpanda Connect - Reads the MySQL binlog using the mysql_cdc input and publishes events to Redpanda.
• MySQL - The source database with binary logging enabled.

Let's start our tutorial by looking at the Docker Compose file. We encourage you to set up a working directory, and follow along.

mkdir cdc-demo && cd cdc-demo

Copy your license file into the working directory:

cp /path/to/redpanda_license.txt

Create a file called docker-compose.yml and paste in the following:

services:
  redpanda:
    image: redpandadata/redpanda:latest
    command:
      - redpanda start
      - --overprovisioned
      - --smp 1
      - --memory 512M
      - --reserve-memory 0M
      - --node-id 0
      - --check=false
      - --kafka-addr PLAINTEXT://0.0.0.0:9092
      - --advertise-kafka-addr PLAINTEXT://redpanda:9092
    ports:
      - "9092:9092"
    healthcheck:
      test: ["CMD", "rpk", "cluster", "info"]
      interval: 5s
      timeout: 5s
      retries: 10

  mysql:
    image: mysql:8.0
    environment:
      MYSQL_ROOT_PASSWORD: rootpw
      MYSQL_USER: mysqluser
      MYSQL_PASSWORD: mysqlpw
      MYSQL_DATABASE: pandashop
    command:
      - --log-bin=mysql-bin
      - --binlog-format=ROW
      - --binlog-row-image=FULL
      - --server-id=1
    volumes:
      - ./mysql-init:/docker-entrypoint-initdb.d
    ports:
      - "3306:3306"
    healthcheck:
      test: ["CMD", "mysqladmin", "ping", "-h", "localhost", "-u", "root", "-prootpw"]
      interval: 5s
      timeout: 5s
      retries: 10

  connect:
    image: redpandadata/connect:latest
    depends_on:
      redpanda:
        condition: service_healthy
      mysql:
        condition: service_healthy
    environment:
      REDPANDA_LICENSE_FILEPATH: /etc/redpanda/redpanda.license
    volumes:
      - ./pipeline.yaml:/etc/redpanda-connect/pipeline.yaml
      - ./redpanda_license.txt:/etc/redpanda/redpanda.license
    entrypoint: sh
    command: -c "mkdir -p /tmp/cdc-checkpoints && /redpanda-connect run /etc/redpanda-connect/pipeline.yaml"

These services are in the same network, so they are reachable from each other. Their service names resolve to their IP addresses in the network.

Some important parameters on the MySQL service:

• --binlog-format=ROW - tells MySQL to log the full before and after value of every changed row. Redpanda Connect requires this.
• --binlog-row-image=FULL - ensures that every column is included in the log entry, even columns that were not changed.
• --server-id=1 - required for MySQL replication. Every server in a replication topology needs a unique ID.

The mysql-init volume mounts a SQL script that runs automatically the first time the container starts. It creates the orders table in a pandashop database and seeds it with a few rows.

Create the directory and the bootstrap file:

mkdir mysql-init

Create a file called mysql-init/bootstrap.sql and paste in the following:

USE pandashop;

GRANT REPLICATION SLAVE, REPLICATION CLIENT, RELOAD ON *.* TO 'mysqluser'@'%';
FLUSH PRIVILEGES;

CREATE TABLE orders (
  id          INT AUTO_INCREMENT PRIMARY KEY,
  customer_id INT NOT NULL,
  status      VARCHAR(50) NOT NULL DEFAULT 'pending',
  total       DECIMAL(10, 2) NOT NULL,
  created_at  TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

INSERT INTO orders (customer_id, status, total) VALUES
  (1, 'pending',    49.99),
  (2, 'shipped',   120.00),
  (3, 'delivered',  75.50);

The GRANT statement gives mysqluser the privileges the mysql_cdc connector needs to read the binlog and take a consistent snapshot.

Now let's look at the Redpanda Connect pipeline. This is where the CDC magic happens.

Create a file called pipeline.yaml in your working directory and paste in the following:

input:
  mysql_cdc:
    dsn: "mysqluser:mysqlpw@tcp(mysql:3306)/pandashop"
    tables:
      - orders
    checkpoint_cache: checkpoint_cache
    stream_snapshot: true
    snapshot_max_batch_size: 1000

pipeline:
  processors:
    - mapping: |
        root       = this
        meta topic = meta("table")

output:
  redpanda:
    seed_brokers:
      - redpanda:9092
    topic: ${! meta("topic") }
    key: ${! json("id") }
    batching:
      count: 100
      period: 1s

cache_resources:
  - label: checkpoint_cache
    file:
      directory: /tmp/cdc-checkpoints

Some important parameters:

• dsn - the connection string to MySQL in user:password@tcp(host:port)/database format.
• tables - the list of tables to track. You can add as many as you need.
• stream_snapshot - when set to true, the pipeline streams all existing rows as read events on first run before switching to live changes. Set it to false if you only want changes going forward.
• checkpoint_cache - after processing each batch, Redpanda Connect writes the current binlog position to disk. If the pipeline restarts, it reads that position and picks up from where it left off.
• topic: ${! meta("topic") } - routes each event to a Redpanda topic named after the source table.
• key: ${! json("id") } - keys each message on the row's id column, which keeps updates to the same row ordered on the same partition.

Now we need to start capturing changes. Start everything up:

docker compose up -d

Wait about ten seconds for the health checks to pass. Then log into MySQL and take a look at the source table:

docker compose exec mysql mysql -u mysqluser -pmysqlpw pandashop

SELECT * FROM orders;

+----+-------------+-----------+--------+---------------------+
| id | customer_id | status    | total  | created_at          |
+----+-------------+-----------+--------+---------------------+
|  1 |           1 | pending   |  49.99 | 2024-01-15 10:00:00 |
|  2 |           2 | shipped   | 120.00 | 2024-01-15 10:00:01 |
|  3 |           3 | delivered |  75.50 | 2024-01-15 10:00:02 |
+----+-------------+-----------+--------+---------------------+

This is your source table. Exit MySQL:

exit

Then check the list of topics in Redpanda:

docker compose exec redpanda rpk topic list

NAME    PARTITIONS  REPLICAS
orders  1           1

As you can see, Redpanda Connect created a topic for the orders table. Now start consuming it:

docker compose exec redpanda rpk topic consume orders

You should see the three seed rows come through as read events from the initial snapshot. This is the representation of a change to the row in the orders table. Each message contains the full row value and metadata about the operation type.

Do not interrupt consumption. Open a second terminal and connect to MySQL to make some changes:

docker compose exec mysql mysql -u mysqluser -pmysqlpw pandashop

Insert a new order:

INSERT INTO orders (customer_id, status, total) VALUES (4, 'pending', 299.00);

Update an existing one:

UPDATE orders SET status = 'shipped' WHERE id = 1;

Delete one:

DELETE FROM orders WHERE id = 3;

As you update, you should see new records arriving in the consumer. Every message includes metadata that tells you exactly what happened:

Check our CDC catalog for more

Now that you’ve seen how easy it is to set up CDC streams with Redpanda, feel free to play around and set them up from other popular database systems.

If you are working with a different database, Redpanda Connect has native CDC inputs for PostgreSQL, MongoDB, Oracle, Microsoft SQL Server, Amazon DynamoDB, Google Cloud Spanner, and TigerBeetle. You can find all of them in the Redpanda Connect CDC components catalog.

And don't forget to join the Redpanda Community on Slack where you can share what you're working on and learn what others in our community are building.