CDC (Change Data Capture)

Change Data Capture connector for streaming database changes in real-time. Captures INSERT, UPDATE, and DELETE operations from the database transaction log.

Supported Databases

Database	Status
MySQL	Supported

Quick Example

apiVersion: laminar.io/v1
kind: Profile
spec:
  name: mysql-cdc
  connector: cdc
  config:
    database_type: mysql
    hostname: localhost
    port: 3306
    username: replication_user
    password: your-password
    server_name: mysql-prod
 
---
apiVersion: laminar.io/v1
kind: Table
spec:
  name: orders_cdc
  connector: cdc
  connection_profile_id: mysql-cdc
  config:
    database_include_list: ecommerce
    table_include_list: ecommerce.orders
    snapshot_mode: initial
    offset_mode: earliest
  schema:
    format:
      json: {}
    fields:
      - field_name: id
        field_type:
          type:
            primitive: Int64
        nullable: false
      - field_name: customer_id
        field_type:
          type:
            primitive: Int64
        nullable: false
      - field_name: total
        field_type:
          type:
            primitive: F64
        nullable: false
      - field_name: status
        field_type:
          type:
            primitive: Utf8
        nullable: false

Configuration

Required

Property	Type	Description
`database_type`	string	Database type (`mysql`)
`hostname`	string	Database server hostname or IP
`username`	string	Database user with replication privileges
`password`	string	Password for the database user
`server_name`	string	Logical server name for identification

Optional

Property	Type	Default	Description
`port`	number	3306	Database port

MySQL User Permissions

The database user needs replication privileges:

GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT
ON *.* TO 'replication_user'@'%';

JSON Schema Reference

Connection Profile Schema

{
  "type": "object",
  "properties": {
    "database_type": {"type": "string", "enum": ["mysql"]},
    "hostname": {"type": "string"},
    "port": {"type": "number", "default": 3306},
    "username": {"type": "string"},
    "password": {"type": "string"},
    "server_name": {"type": "string"}
  },
  "required": ["database_type", "hostname", "username", "password", "server_name"]
}

Connection Table Schema

{
  "type": "object",
  "properties": {
    "database_include_list": {"type": "string"},
    "table_include_list": {"type": "string"},
    "snapshot_mode": {"type": "string", "enum": ["initial", "schema_only"]},
    "offset_mode": {"type": "string", "enum": ["earliest"]}
  },
  "required": ["database_include_list", "table_include_list"]
}