Connect

MySQL CDC Patterns

The mysql_cdc input captures row-level changes from MySQL tables using the binary log (binlog). Use these patterns to filter, transform, and route MySQL CDC events to Redpanda, S3, and other destinations, and to plan captures that span many tables.

Use this cookbook to:

Apply reusable patterns for capturing MySQL CDC events
Adapt integration patterns to route CDC data to Redpanda and S3
Identify patterns for filtering and transforming change events
Plan inputs and pipelines that capture changes from many MySQL tables

Prerequisites

Before using these patterns, configure the following.

Redpanda CLI

Install the Redpanda CLI (rpk) to run Redpanda Connect. See rpk installation for installation instructions.

MySQL binlog

The source MySQL database must have binary logging enabled with row-based replication:

-- Verify binary logging is enabled
SHOW VARIABLES LIKE 'log_bin';

-- Verify row-based format
SHOW VARIABLES LIKE 'binlog_format';

Set log_bin to ON and binlog_format to ROW. For cloud-managed MySQL, see:

Checkpoint cache

The mysql_cdc input requires a cache to store the binlog position between restarts. The examples in this cookbook use a Redis cache:

export REDIS_URL=redis://localhost:6379

For production use, any persistent cache supported by Redpanda Connect works, such as aws_dynamodb or postgres.

Environment variables

The examples in this cookbook use environment variables for configuration:

export MYSQL_DSN="user:password@tcp(localhost:3306)/mydb" (1)
export MYSQL_TABLES="orders,customers" (2)
export REDIS_URL=redis://localhost:6379 (3)
export REDPANDA_BROKERS=localhost:9092 (4)
export S3_BUCKET=cdc-archive (5)

1	The MySQL DSN in `user:password@tcp(host:port)/database` format.
2	Comma-separated list of tables to capture. List table names without a database prefix: the input reads tables from the database named in the DSN.
3	The Redis URL for checkpoint storage.
4	The Redpanda broker addresses (for Redpanda output examples).
5	The S3 bucket name (for S3 output examples).

Capture CDC events

The simplest pattern captures all change events from MySQL tables and outputs them with metadata:

input:
  mysql_cdc:
    dsn: ${MYSQL_DSN}
    tables:
      - orders
    checkpoint_cache: redis_cache
    stream_snapshot: true

pipeline:
  processors:
    - mapping: |
        root.operation = meta("operation")
        root.table = meta("table")
        root.binlog_position = meta("binlog_position")
        root.data = this
        root.timestamp = now()

output:
  stdout:
    codec: lines

cache_resources:
  - label: redis_cache
    redis:
      url: ${REDIS_URL}

For details on the CDC event message structure and available metadata fields, see the metadata section in the connector reference.

Filter CDC events

Filter events to process only specific change types using the operation metadata field:

input:
  mysql_cdc:
    dsn: ${MYSQL_DSN}
    tables:
      - orders
    checkpoint_cache: redis_cache
    stream_snapshot: true

pipeline:
  processors:
    - mapping: |
        # Drop delete events, pass through inserts and updates
        root = if meta("operation") == "delete" {
          deleted()
        } else {
          {
            "operation": meta("operation"),
            "table": meta("table"),
            "data": this,
            "timestamp": now()
          }
        }

output:
  stdout:
    codec: lines

cache_resources:
  - label: redis_cache
    redis:
      url: ${REDIS_URL}

This pattern:

Filters out delete events, passing through insert and update operations
Transforms the event to a simplified format with a timestamp

Route to Redpanda

Stream MySQL changes to Redpanda for real-time processing:

input:
  mysql_cdc:
    dsn: ${MYSQL_DSN}
    tables:
      - orders
      - customers
    checkpoint_cache: redis_cache
    stream_snapshot: true

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

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

cache_resources:
  - label: redis_cache
    redis:
      url: ${REDIS_URL}

This pattern:

Uses the table name as the Redpanda topic
Batches messages for efficient delivery
Sets the message key to the row’s primary key field (update json("id") to match your table’s primary key column)

Route to S3

Archive CDC events to S3 for long-term storage and analytics:

input:
  mysql_cdc:
    dsn: ${MYSQL_DSN}
    tables:
      - orders
    checkpoint_cache: redis_cache
    stream_snapshot: true

pipeline:
  processors:
    - mapping: |
        root.operation = meta("operation")
        root.table = meta("table")
        root.data = this
        root.timestamp = now()

output:
  aws_s3:
    bucket: ${S3_BUCKET}
    path: >-
      cdc/${! meta("table") }/${! timestamp_unix().format_timestamp("2006/01/02/15") }/${! uuid_v4() }.ndjson
    batching:
      count: 1000
      period: 5m
      processors:
        - archive:
            format: lines

cache_resources:
  - label: redis_cache
    redis:
      url: ${REDIS_URL}

This pattern:

Organizes files by table and time-based partitions (year/month/day/hour)
Batches events and archives them as newline-delimited JSON
Uses UUID file names to prevent collisions

Route by event type

Route different event types to different destinations:

input:
  mysql_cdc:
    dsn: ${MYSQL_DSN}
    tables:
      - orders
    checkpoint_cache: redis_cache
    stream_snapshot: true

output:
  switch:
    cases:
      - check: meta("operation") == "insert"
        output:
          redpanda:
            seed_brokers:
              - ${REDPANDA_BROKERS}
            topic: mysql.orders.inserts
      - output:
          redpanda:
            seed_brokers:
              - ${REDPANDA_BROKERS}
            topic: mysql.orders.changes

cache_resources:
  - label: redis_cache
    redis:
      url: ${REDIS_URL}

This pattern:

Routes insert events to one Redpanda topic and all other change events to another
Supports specialized downstream consumers per operation type

Configure replication mode

The mysql_cdc input supports two replication modes controlled by the stream_snapshot field:

stream_snapshot: true: Captures a full snapshot of existing table data before streaming live changes. Use this when you need a complete initial load.
stream_snapshot: false: Skips the snapshot and streams only changes from the current binlog position. Use this when you only need new changes going forward.

input:
  mysql_cdc:
    dsn: ${MYSQL_DSN}
    tables:
      - orders
    checkpoint_cache: redis_cache
    stream_snapshot: true
    snapshot_max_batch_size: 1000 (1)

cache_resources:
  - label: redis_cache
    redis:
      url: ${REDIS_URL}

1	Number of rows to read per batch during snapshot processing.

Capture many tables

To capture changes from dozens or hundreds of tables, group many tables into each mysql_cdc input instead of creating one input or pipeline for each table. Fewer inputs mean fewer binlog connections to your MySQL server and fewer pipelines to run.

How inputs read the binlog

MySQL writes changes for every table on the server to a single binary log. Each mysql_cdc input opens its own binlog connection, and the server sends that connection the full binlog stream. The input then discards events for tables that are not in its tables list.

As a result, every additional input reads and transfers the same binlog again. If you create one pipeline per table for 100 tables, the server streams the full binlog 100 times, and each pipeline discards almost all of it. A single input that lists all 100 tables reads the binlog once.

Group tables into lanes

A lane (sometimes called a swimlane) is one mysql_cdc input that reads the binlog once for a group of tables. Start with as few lanes as possible, and add a lane only when you have a reason to separate tables:

Tables in different databases. Each input reads tables from the database named in its dsn, so tables in another database need their own input.
Very large tables whose snapshot you want to run, restart, or tune separately from the rest. See Plan snapshots for large lanes.
Groups of tables that you want to deploy, stop, or change independently, such as tables owned by different teams.

Each lane stores its own binlog position. If lanes share a checkpoint cache, give each lane a unique checkpoint_key. Lanes that use the same key overwrite each other’s position. The mysql_cdc input has no setting for a replica server ID.

The following lane captures six tables from one database and writes each table to its own Redpanda topic:

input:
  mysql_cdc:
    dsn: ${MYSQL_DSN}
    tables: (1)
      - orders
      - order_items
      - customers
      - addresses
      - products
      - inventory
    checkpoint_cache: redis_cache
    checkpoint_key: mysql_binlog_position_sales (2)
    stream_snapshot: true
    max_parallel_snapshot_tables: 4 (3)

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

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

cache_resources:
  - label: redis_cache
    redis:
      url: ${REDIS_URL}

1	List every table in the lane. Use table names only, without a database prefix.
2	A checkpoint key that is unique to this lane.
3	The number of tables to snapshot in parallel. The default is `1`.

The mapping sets the topic from the table metadata field, so one output serves every table in the lane.

Plan snapshots for large lanes

When stream_snapshot is true and the checkpoint cache holds no binlog position for the lane, the input snapshots every table in its tables list before it streams changes. Keep the following behavior in mind when you size a lane:

The input runs FLUSH TABLES … WITH READ LOCK on all tables in the lane, which blocks writes to them until the input has opened its snapshot transactions and read the binlog position.
The input stores the binlog position only after the whole snapshot is acknowledged. If the pipeline restarts before then, the snapshot of every table in the lane starts again.
After the snapshot, the input streams changes from the binlog position that it read before the snapshot. Your MySQL server must still have the binlog file for that position when the snapshot finishes, so set the binlog retention period on the server to cover the full snapshot time.
Use max_parallel_snapshot_tables to snapshot several tables at the same time. Each parallel snapshot worker holds its own transaction on a separate MySQL connection.

Add tables to an existing lane

If you add a table to the tables list of a lane that already has a stored binlog position, the input does not snapshot the new table. It streams changes to that table from the stored position onward, but it does not read rows that existed before.

To load the existing rows of a new table, capture it in a new lane with its own checkpoint_key and stream_snapshot: true, instead of adding it to an existing lane.

Pipeline limits in Redpanda Cloud

In Redpanda Cloud, each cluster can create up to 100 pipelines. To run more, contact Redpanda support. A design with one pipeline per table can reach this limit quickly. For details, see Redpanda Connect quickstart.

Planning checklist

Before you build pipelines for a large migration, answer these questions:

How many tables do you need to capture, and in how many databases?
How many lanes do you need? Start with one lane per database, and split only for the reasons in Group tables into lanes.
Does each lane have a unique checkpoint_key, or its own checkpoint cache?
How many rows does each lane need to snapshot, and how long does each snapshot take?
Does the binlog retention period on your MySQL server cover the longest snapshot?
Do all tables that you snapshot have a primary key? Snapshots require one.
How often do you plan to add tables, and how do you load their existing rows?
In Redpanda Cloud, how many pipelines does the cluster run in total, including pipelines for other workloads?

Troubleshoot common issues

Use these steps to diagnose and fix the most common problems with the mysql_cdc input.

No events received

If you’re not receiving events:

Verify binary logging is enabled:

SHOW VARIABLES LIKE 'log_bin';
SHOW VARIABLES LIKE 'binlog_format';

Confirm the user has the required MySQL privileges:

GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'your_user'@'%';
GRANT SELECT ON mydb.* TO 'your_user'@'%';

Check that the names in tables are table names only, such as orders, not mydb.orders. The input reads tables from the database named in dsn.

Pipeline restarts lose position

If the pipeline restarts and replays events from the beginning:

Verify the checkpoint cache is persistent and accessible.
Check that checkpoint_key is consistent across restarts (default: mysql_binlog_position).
Use a durable cache backend such as Redis with persistence enabled, aws_dynamodb, or postgres.

Duplicate events

The mysql_cdc input provides at-least-once delivery. If the pipeline fails between checkpoints, events may be re-read on restart. To handle duplicates:

Use idempotent processing in downstream systems.
Deduplicate using the binlog_position metadata field.
Lower checkpoint_limit to reduce the window of possible duplicates.

Next steps

Suggested labs

Stream Jira Issues to Redpanda for Real-Time Metrics

Search all labs

Was this helpful?

group Ask in the community

mail Share your feedback

What do you think of this page?

Let us know more:

Let us contact you about your feedback:

MySQL CDC Patterns

Prerequisites

Redpanda CLI

MySQL binlog

Checkpoint cache

Environment variables

Capture CDC events

Filter CDC events

Route to Redpanda

Route to S3

Route by event type

Configure replication mode

Capture many tables

How inputs read the binlog

Group tables into lanes

Plan snapshots for large lanes

Add tables to an existing lane

Pipeline limits in Redpanda Cloud

Planning checklist

Troubleshoot common issues

No events received

Pipeline restarts lose position

Duplicate events

Next steps

Suggested labs