MySQL CDC

​

Overview

​

Configuration

​

Required Parameters

​

Optional Parameters

​

SSL Configuration

​

Record Formats

​

Standard Event Format

{
  "op": "c",
  "schema": "mydb",
  "table": "users",
  "ts": 1640995200,
  "tx": {
    "gtid": "uuid:1-100",
    "xid": 12345
  },
  "before": {
    "id": 1,
    "name": "John",
    "email": "john@example.com"
  },
  "after": {
    "id": 1,
    "name": "John",
    "email": "john@example.com"
  },
  "offset": {
    "file": "mysql-bin.000001",
    "pos": 1234,
    "gtid": "uuid:1-100"
  }
}

​

DB Records Only Format

{
  "data": {
    "id": 1,
    "name": "John",
    "email": "john@example.com",
    "_mage_created_at": "2021-12-31T23:00:00+00:00",
    "_mage_updated_at": null,
    "_mage_deleted_at": null
  },
  "metadata": {
    "schema": "mydb",
    "table": "users",
    "key_columns": ["id"],
    "operation": "INSERT"
  }
}

​

Event Types

​

Primary Key Detection

1. Schema Discovery: Queries INFORMATION_SCHEMA for table structure
2. Primary Key Detection: Identifies actual primary key columns from database constraints
3. Caching: Stores schema and primary key information for performance
4. Key Extraction: Extracts primary key column names from row events

{
  "metadata": {
    "key_columns": ["user_id", "tenant_id"]
  }
}

​

Examples

connector_type: mysql
host: localhost
port: 3306
user: repl_user
password: password
databases: ["mydb"]
return_db_records_only: true

​

Prerequisites

​

MySQL Server Configuration

-- Enable binary logging
SET GLOBAL log_bin = ON;

-- Set binlog format to ROW
SET GLOBAL binlog_format = 'ROW';

-- Enable GTID (recommended)
SET GLOBAL gtid_mode = ON;
SET GLOBAL enforce_gtid_consistency = ON;

​

User Permissions

-- Create replication user
CREATE USER 'repl_user'@'%' IDENTIFIED BY 'password';

-- Grant replication privileges
GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'repl_user'@'%';

-- Grant SELECT privileges for schema queries
GRANT SELECT ON information_schema.* TO 'repl_user'@'%';
GRANT SELECT ON mydb.* TO 'repl_user'@'%';

​

Troubleshooting

import logging
logging.basicConfig(level=logging.DEBUG)

​

Integration with Generic IO Sink

• Automatic Column Type Mapping: Automatically maps _mage_* timestamp columns to appropriate database types (TIMESTAMP, DATETIME2, DateTime64, etc.) based on the target database
• Metadata Interpolation: Use metadata values from MySQL CDC events in sink configurations for dynamic routing and table naming

​

Supported Databases

• BigQuery
• ClickHouse
• DuckDB
• MySQL
• MSSQL
• Postgres

​

Metadata Interpolation

• {schema}: Database/schema name from the event
• {table}: Table name from the event
• {key_columns}: List of primary key column names (e.g., ["id"] or ["user_id", "tenant_id"])

​

Example Configurations

connector_type: bigquery
profile: default
config:
  table_id: "{schema}.{table}"  # Dynamic table names from metadata
  unique_conflict_method: UPDATE
  unique_constraints: "{key_columns}"  # Uses primary keys from metadata

​

How Metadata Interpolation Works

1. Message Grouping: Messages are automatically grouped by their interpolated config values. For example, if you use table_name: "{schema}_{table}", messages from mydb.users will be grouped together and written to mydb_users table.
2. Key Columns Interpolation: When using {key_columns} in unique_constraints, the sink automatically:
   • Converts the list to a string representation during interpolation
   • Parses it back to a list (supports both "['id']" and '["id"]' formats)
   • Uses it for upsert operations based on unique_conflict_method

​

Best Practices

1. Use GTID: Enable GTID for reliable resume capabilities
2. Filter Events: Disable unnecessary event types for performance
3. Monitor Resources: Watch memory and CPU usage
4. Test Resume: Verify checkpoint functionality
5. Secure Connections: Use SSL in production
6. Regular Backups: Backup checkpoint files
7. Schema Validation: Test with schema changes
8. Use Generic IO Sink: Leverage automatic type mapping and metadata interpolation for flexible data routing
9. Metadata Interpolation: Use {schema}, {table}, and {key_columns} for dynamic table routing and upsert configuration

​

Timestamp Handling

• _mage_created_at: Set to event timestamp (datetime) for INSERT operations
• _mage_updated_at: Set to event timestamp (datetime) for UPDATE operations
• _mage_deleted_at: Set to event timestamp (datetime) for DELETE operations

• Proper type handling in downstream databases
• Automatic type conversion in Generic IO sink
• Consistent timezone handling across systems

​

Limitations

• MySQL 5.7+: Requires MySQL 5.7 or later
• ROW Format: Requires binlog_format = ROW
• Network Dependency: Requires stable network connection
• Memory Usage: Schema caching uses memory
• Binlog Retention: Depends on MySQL binlog retention settings
• Timezone: All timestamps are in UTC timezone