# MySQL CDC Patterns

> For the complete documentation index, see [llms.txt](https://docs.redpanda.com/llms.txt). Component-specific: [connect-full.txt](https://docs.redpanda.com/connect-full.txt)

---
title: MySQL CDC Patterns
latest-connect-version: 4.105.0
latest-operator-version: v26.2.1
latest-console-tag: v3.10.0
latest-redpanda-tag: v26.2.1
docname: mysql_cdc
page-component-name: connect
page-version: master
page-component-version: master
page-component-title: Connect
page-relative-src-path: mysql_cdc.adoc
page-edit-url: https://github.com/redpanda-data/rp-connect-docs/edit/main/modules/cookbooks/pages/mysql_cdc.adoc
description: Learn how to capture, filter, transform, and route MySQL change data capture (CDC) events with Redpanda Connect.
page-topic-type: cookbook
personas: streaming_developer, data_engineer
learning-objective-1: Apply reusable patterns for capturing MySQL CDC events
learning-objective-2: Adapt integration patterns to route CDC data to Redpanda and S3
learning-objective-3: Identify patterns for filtering and transforming change events
page-git-created-date: "2026-06-17"
page-git-modified-date: "2026-06-17"
---

<!-- Source: https://docs.redpanda.com/connect/cookbooks/mysql_cdc.md -->

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.

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


## [](#prerequisites)Prerequisites

Before using these patterns, configure the following.

### [](#redpanda-cli)Redpanda CLI

Install the Redpanda CLI (`rpk`) to run Redpanda Connect. See [rpk installation](https://docs.redpanda.com/connect/get-started/quickstarts/rpk/) for installation instructions.

### [](#mysql-binlog)MySQL binlog

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

```sql
-- 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:

-   [Change data capture on Amazon RDS for MySQL](https://aws.amazon.com/blogs/database/enable-change-data-capture-on-amazon-rds-for-mysql-applications-that-are-using-xa-transactions/)

-   [Azure MySQL Database CDC](https://learn.microsoft.com/en-us/fabric/real-time-hub/add-source-mysql-database-cdc)

-   [Google Cloud SQL for MySQL](https://cloud.google.com/datastream/docs/configure-cloudsql-mysql)


### [](#checkpoint-cache)Checkpoint cache

The `mysql_cdc` input requires a [cache](https://docs.redpanda.com/connect/components/caches/about/) to store the binlog position between restarts. The examples in this cookbook use a Redis cache:

```bash
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)Environment variables

The examples in this cookbook use environment variables for configuration:

```bash
export MYSQL_DSN="user:password@tcp(localhost:3306)/mydb" (1)
export MYSQL_TABLES="mydb.orders,mydb.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 in database.table format. |
| 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)Capture CDC events

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

```yaml
input:
  mysql_cdc:
    dsn: ${MYSQL_DSN}
    tables:
      - mydb.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](https://docs.redpanda.com/connect/components/inputs/mysql_cdc/#_metadata) section in the connector reference.

## [](#filter-cdc-events)Filter CDC events

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

```yaml
input:
  mysql_cdc:
    dsn: ${MYSQL_DSN}
    tables:
      - mydb.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)Route to Redpanda

Stream MySQL changes to Redpanda for real-time processing:

```yaml
input:
  mysql_cdc:
    dsn: ${MYSQL_DSN}
    tables:
      - mydb.orders
      - mydb.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)Route to S3

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

```yaml
input:
  mysql_cdc:
    dsn: ${MYSQL_DSN}
    tables:
      - mydb.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 by event type

Route different event types to different destinations:

```yaml
input:
  mysql_cdc:
    dsn: ${MYSQL_DSN}
    tables:
      - mydb.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)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.


```yaml
input:
  mysql_cdc:
    dsn: ${MYSQL_DSN}
    tables:
      - mydb.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. |
| --- | --- |

## [](#troubleshoot-common-issues)Troubleshoot common issues

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

### [](#no-events-received)No events received

If you’re not receiving events:

1.  Verify binary logging is enabled:

    ```sql
    SHOW VARIABLES LIKE 'log_bin';
    SHOW VARIABLES LIKE 'binlog_format';
    ```

2.  Confirm the user has the required MySQL privileges:

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

3.  Check that the table names in `tables` use the `database.table` format.


### [](#pipeline-restarts-lose-position)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)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)Next steps

-   [MySQL CDC input reference](https://docs.redpanda.com/connect/components/inputs/mysql_cdc/)

-   [Redis cache](https://docs.redpanda.com/connect/components/caches/redis/)

-   [Redpanda output](https://docs.redpanda.com/connect/components/outputs/redpanda/)


## Suggested labs

-   [Stream Jira Issues to Redpanda for Real-Time Metrics](https://docs.redpanda.com/labs/docker-compose/jira-metrics-pipeline/)

[Search all labs](https://docs.redpanda.com/labs)