import ChangeLog from ‘../changelog/connector-cdc-mysql.md’;

MySQL CDC

MySQL CDC source 连接器

支持这些引擎

SeaTunnel Zeta
Flink

描述

MySQL CDC连接器允许从MySQL数据库读取快照和增量数据. 本文档描述了如何配置MySQL CDC连接器以对MySQL数据库运行SQL查询.

主要功能

支持的数据源信息

数据源	支持的版本	Driver	Url	Maven
MySQL	MySQL: 5.5, 5.6, 5.7, 8.0.x RDS MySQL: 5.6, 5.7, 8.0.x	com.mysql.cj.jdbc.Driver	jdbc:mysql://localhost:3306/test	https://mvnrepository.com/artifact/mysql/mysql-connector-java/8.0.28

依赖

安装Jdbc驱动

对于Flink引擎

你需要确保 jdbc 驱动 jar package 已经放在目录 ${SEATUNNEL_HOME}/plugins/.

对于SeaTunnel Zeta引擎

你需要确保 jdbc 驱动 jar package 已经放在目录 ${SEATUNNEL_HOME}/lib/.

创建MySQL用户

你必须定义一个MySQL用户，该用户对Debezium MySQL连接器所监控的所有数据库拥有适当的权限.

创建MySQL用户:

mysql> CREATE USER 'user'@'localhost' IDENTIFIED BY 'password';

给用户赋予所需权限:

mysql> GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'user' IDENTIFIED BY 'password';

最终确定用户权限:

mysql> FLUSH PRIVILEGES;

启用MySQL Binlog

一定要为MySQL复制启用binlog。binlog记录事务更新以供复制工具传播更改.

检查log-bin是否已经设置为on:

mysql> show variables where variable_name in ('log_bin', 'binlog_format', 'binlog_row_image', 'gtid_mode', 'enforce_gtid_consistency');
+--------------------------+----------------+
| Variable_name            | Value          |
+--------------------------+----------------+
| binlog_format            | ROW            |
| binlog_row_image         | FULL           |
| enforce_gtid_consistency | ON             |
| gtid_mode                | ON             |
| log_bin                  | ON             |
+--------------------------+----------------+

如果log_bin的值不是on, 配置你的MySQL server配置文件($MYSQL_HOME/mysql.cnf)，配置文件中包含以下属性，这些属性在以下表格中有描述:

# Enable binary replication log and set the prefix, expiration, and log format.
# The prefix is arbitrary, expiration can be short for integration tests but would
# be longer on a production system. Row-level info is required for ingest to work.
# Server ID is required, but this will vary on production systems
server-id         = 223344
log_bin           = mysql-bin
expire_logs_days  = 10
binlog_format     = row
# mysql 5.6+ requires binlog_row_image to be set to FULL
binlog_row_image  = FULL

# optional enable gtid mode
# mysql 5.6+ requires gtid_mode to be set to ON, but not required by mysql 8.0+
gtid_mode = on
enforce_gtid_consistency = on

重启MySQL Server

/etc/inint.d/mysqld restart

修改之后再检查一次binlog的状态:

MySQL 5.5:

mysql> show variables where variable_name in ('log_bin', 'binlog_format', 'binlog_row_image', 'gtid_mode', 'enforce_gtid_consistency');
+--------------------------+----------------+
| Variable_name            | Value          |
+--------------------------+----------------+
| binlog_format            | ROW            |
| log_bin                  | ON             |
+--------------------------+----------------+

MySQL 5.6+:

mysql> show variables where variable_name in ('log_bin', 'binlog_format', 'binlog_row_image', 'gtid_mode', 'enforce_gtid_consistency');
+--------------------------+----------------+
| Variable_name            | Value          |
+--------------------------+----------------+
| binlog_format            | ROW            |
| binlog_row_image         | FULL           |
| enforce_gtid_consistency | ON             |
| gtid_mode                | ON             |
| log_bin                  | ON             |
+--------------------------+----------------+

MySQL 8.0+:

show variables where variable_name in ('log_bin', 'binlog_format', 'binlog_row_image', 'gtid_mode', 'enforce_gtid_consistency')
+--------------------------+----------------+
| Variable_name            | Value          |
+--------------------------+----------------+
| binlog_format            | ROW            |
| binlog_row_image         | FULL           |
| enforce_gtid_consistency | OFF            |
| gtid_mode                | OFF            |
| log_bin                  | ON             |
+--------------------------+----------------+

提示

配置MySQL session超时时长

当为大型数据库初始一致快照时，已建立的连接可能在读取表时超时。可以通过在MySQL配置文件中配置interactive_timeout（交互超时时间）和wait_timeout（等待超时时间）来防止这种行为.

interactive_timeout: 服务器在关闭交互连接之前等待活动（交互操作）的秒数. 详见 MySQL’s documentation.
wait_timeout: 服务器在关闭非交互式连接之前等待其活动的秒数. 详见 MySQL’s documentation.

更多的数据库配置，见 Debezium MySQL Connector

数据类型映射

Mysql数据类型	SeaTunnel数据类型
BIT(1) TINYINT(1)	BOOLEAN
TINYINT	TINYINT
TINYINT UNSIGNED SMALLINT	SMALLINT
SMALLINT UNSIGNED MEDIUMINT MEDIUMINT UNSIGNED INT INTEGER YEAR	INT
INT UNSIGNED INTEGER UNSIGNED BIGINT	BIGINT
BIGINT UNSIGNED	DECIMAL(20,0)
DECIMAL(p, s) DECIMAL(p, s) UNSIGNED NUMERIC(p, s) NUMERIC(p, s) UNSIGNED	DECIMAL(p,s)
FLOAT FLOAT UNSIGNED	FLOAT
DOUBLE DOUBLE UNSIGNED REAL REAL UNSIGNED	DOUBLE
CHAR VARCHAR TINYTEXT MEDIUMTEXT TEXT LONGTEXT ENUM JSON ENUM	STRING
DATE	DATE
TIME(s)	TIME(s)
DATETIME TIMESTAMP(s)	TIMESTAMP(s)
BINARY VARBINAR BIT(p) TINYBLOB MEDIUMBLOB BLOB LONGBLOB GEOMETRY	BYTES

配置参数选项

参数名称	类型	是否必须	默认值	描述
url	String	是	-	JDBC连接的URL. 例如: `jdbc:mysql://localhost:3306/test`.
username	String	是	-	用来连接到数据库服务的数据库名称.
password	String	是	-	连接到数据库服务所使用的密码.
database-names	List	否	-	要监控的数据库名称.
database-pattern	String	否	.*	要捕获的数据库名称的正则表达式, 例如: `database_prefix.*`.
table-names	List	是	-	要监控的表名. 表名需要包括库名, 例如: `database_name.table_name`
table-pattern	String	是	-	要捕获的表名称的正则表达式. 表名需要包括库名, 例如: `database.\\.table_.`
table-names-config	List	否	-	表配置的列表集合. 例如: [{“table”: “db1.schema1.table1”,“primaryKeys”: [“key1”],“snapshotSplitColumn”: “key2”}]. snapshotSplitColumn 选项必须配置为唯一键(主键或唯一索引). 如果指定了非唯一列，该配置将被忽略，SeaTunnel 会在内部自动选择合适的拆分列.
startup.mode	Enum	否	INITIAL	MySQL CDC 消费者的可选启动模式, 有效枚举值为 `initial`, `earliest`, `latest` , `specific` 和 `timestamp`. `initial`: 启动时同步历史数据, 然后同步增量数据. `earliest`: 从尽可能最早的偏移量开始启动. `latest`: 从最近的偏移量启动. `specific`: 从用户提供的特定偏移量开始启动. `timestamp`: 从用户提供的特定时间戳开始启动.
startup.specific-offset.file	String	否	-	从指定的binlog日志文件名开始. 注意, 当使用 `startup.mode` 选项为 `specific` 时，此选项为必填项.
startup.specific-offset.pos	Long	否	-	从指定的binlog日志文件位置开始. 注意, 当使用 `startup.mode` 选项为 `specific` 时，此选项为必填项.
startup.timestamp	Long	No	-	从指定的binlog时间戳文件位置开始. 注意, 当使用 `startup.mode` 选项为 `timestamp` 时，此选项为必填项.
stop.mode	Enum	否	NEVER	MySQL CDC 消费者的可选停止模式, 有效枚举值为 `never`, `latest` 和 `specific`. `never`: 实时任务一直运行不停止. `latest`: 从最新的偏移量处停止. `specific`: 从用户提供的特定偏移量处停止.
stop.specific-offset.file	String	否	-	从指定的binlog日志文件名停止. 注意, 当使用 `stop.mode` 选项为 `specific` 时，此选项为必填项.
stop.specific-offset.pos	Long	否	-	从指定的binlog日志文件位置停止. 注意, 当使用 `stop.mode` 选项为 `specific` 时，此选项为必填项.
snapshot.split.size	Integer	否	8096	表快照的分割大小（行数）,读取表的快照时,被捕获的表会被分割成多个分割块.
snapshot.fetch.size	Integer	否	1024	每次轮询读取表快照时的最大获取大小.
server-id	String	否	-	此数据库客户端的数字 ID 或数字 ID 范围, 数字 ID 的语法如 `5400`, 数字 ID 范围的语法如 ‘5400-5408’. 每个 ID 在 MySQL 集群中所有当前正在运行的数据库进程里必须是唯一的. 此连接加入 MySQL服务以另外一个服务的身份 (带有此唯一 ID) 以便于能够读取binlog. 默认情况下, 会生成一个介于 6500 到 2,148,492,146 之间的数字, 然而我们建议设置一个明确的值.
server-time-zone	String	否	UTC	数据库服务中的会话时区. 如果没设置, 使用 ZoneId.systemDefault() 来确定服务的时区.
connect.timeout.ms	Duration	否	30000	连接器在尝试连接数据库服务器后，在超时之前应等待的最长时间.
connect.max-retries	Integer	否	3	连接器在构建数据库服务器连接时应重试的最大重试次数.
connection.pool.size	Integer	否	20	jdbc连接池大小.
chunk-key.even-distribution.factor.upper-bound	Double	否	100	块键分布因子的上限. 该因子用于确定表数据是否分布均匀. 如果分布式因子计算结果小于或等于此上限 (即., (MAX(id) - MIN(id) + 1) / row count), 表的分块将被优化以实现均匀分布. 否则, 如果分布因子大于此上限, 该表将被视为分布不均, 并且如果估计的分片数量超过 `sample-sharding.threshold` 所指定的值, 则将使用基于采样的分片策略. 默认值是100.0.
chunk-key.even-distribution.factor.lower-bound	Double	否	0.05	块键分布因子的下限. 该因子用于确定表数据是否分布均匀. 如果计算得出的分布因子大于或等于此下限 (即., (MAX(id) - MIN(id) + 1) / row count), 表的分块将被优化以实现均匀分布. 否则, 如果分布因子小于此下限, 该表将被视为分布不均, 并且如果预估的分片数量超过了 `sample-sharding.threshold` 所指定的值，则将使用基于采样的分片策略. 默认值是 0.05.
sample-sharding.threshold	Integer	否	1000	此配置指定了触发采样分片策略的预估分片数量阈值. 当分配因子超出由 `chunk-key.even-distribution.factor.upper-bound` 和 `chunk-key.even-distribution.factor.lower-bound` 所指定的范围时, 如果估计的分片数量 (按近似行数/块大小计算) 超过此阈值, 则将使用样本分片策略. 这有助于更高效地处理大型数据集. 默认值为 1000 分片.
inverse-sampling.rate	Integer	否	1000	采样分片策略中使用的采样率的倒数. 例如, 如果该值设置为 1000, 则表示在采样过程中应用了 1/1000 的采样率. 此选项在控制采样的粒度方面提供了灵活性, 从而影响最终的分片数量. 在处理非常大的数据集时非常有用, 因为此时更倾向于使用较低的采样率. 默认值为 1000.
split.allow-sampling	Boolean	否	true	是否允许基于采样的分片策略。当设置为 false 时，无论预估分片数是否超过阈值，系统都将回退到非均匀分片方式（迭代查询方式）。默认值为 true。
exactly_once	Boolean	否	false	启用精确一次语义.
format	Enum	否	DEFAULT	MySQL CDC 的可选输出格式, 有效的枚举值为 `DEFAULT`、`COMPATIBLE_DEBEZIUM_JSON`.
schema-changes.enabled	Boolean	否	false	模式演进默认是禁用的. 当前我们只支持 `add column`、`drop column`、`rename column` 和 `modify column`.
schema-changes.include	List	否	-	仅向下游发送列出的 schema change 事件类型（需 `schema-changes.enabled = true`）。为空表示全部允许。详见 Schema change 事件过滤。
schema-changes.exclude	List	否	-	此处列出的 schema change 事件类型不会发送到下游。在 `schema-changes.include` 之后应用；冲突时 exclude 优先。详见 Schema change 事件过滤。
debezium	Config	否	-	传递 Debezium的属性给Debezium嵌入式引擎, 该引擎用于捕获 MySQL 服务的数据变更.
int_type_narrowing	Boolean	否	true	Int类型收窄，如果为 true，则 tinyint(1) 类型将被收窄为 boolean 类型（如果没有精度损失）。目前仅支持 MySQL。
common-options		否	-	Source插件通用参数, 详见 Source Common Options

int_type_narrowing

Int类型收窄，如果为 true，则 tinyint(1) 类型将被收窄为 boolean 类型（如果没有精度损失）。目前仅支持 MySQL。

例：

int_type_narrowing = true

MySQL	SeaTunnel
TINYINT(1)	Boolean

int_type_narrowing = false

MySQL	SeaTunnel
TINYINT(1)	TINYINT

任务示例

简单的示例

支持多表读取

env {
  parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 10000
}

source {
  MySQL-CDC {
    url = "jdbc:mysql://localhost:3306/testdb"
    username = "root"
    password = "root@123"
    table-names = ["testdb.table1", "testdb.table2"]
    
    startup.mode = "initial"
  }
}

sink {
  Console {
  }
}

支持向Kafka发送与Debezium兼容的格式

一定是使用kafka作为sink, 详见 compatible debezium format

支持表的自定义主键

env {
  parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 10000
}

source {
  MySQL-CDC {
    url = "jdbc:mysql://localhost:3306/testdb"
    username = "root"
    password = "root@123"
    
    table-names = ["testdb.table1", "testdb.table2"]
    table-names-config = [
      {
        table = "testdb.table2"
        primaryKeys = ["id"]
      }
    ]
  }
}

sink {
  Console {
  }
}

支持模式演变（表结构变更）

env {
  # You can set engine configuration here
  parallelism = 5
  job.mode = "STREAMING"
  checkpoint.interval = 5000
  read_limit.bytes_per_second=7000000
  read_limit.rows_per_second=400
}

source {
  MySQL-CDC {
    server-id = 5652-5657
    username = "st_user_source"
    password = "mysqlpw"
    table-names = ["shop.products"]
    url = "jdbc:mysql://mysql_cdc_e2e:3306/shop"
    
    schema-changes.enabled = true
  }
}

sink {
  jdbc {
    url = "jdbc:mysql://mysql_cdc_e2e:3306/shop"
    driver = "com.mysql.cj.jdbc.Driver"
    user = "st_user_sink"
    password = "mysqlpw"
    generate_sink_sql = true
    database = shop
    table = mysql_cdc_e2e_sink_table_with_schema_change_exactly_once
    primary_keys = ["id"]
    is_exactly_once = true
    xa_data_source_class_name = "com.mysql.cj.jdbc.MysqlXADataSource"
  }
}

Schema change 事件过滤

当 schema-changes.enabled = true 时，可通过 schema-changes.include / schema-changes.exclude 进一步控制哪些 schema change 事件类型会被发送到下游。

使用以下 SeaTunnel 统一的规范名称

规范名称	操作
`add.column`	新增列
`drop.column`	删除列
`modify.column`	修改列的类型/属性，列名不变
`change.column`	列重命名，可同时改类型
`update.columns`	上述四种列级变更的分组别名

优先级规则（确定性）：

若设置了 schema-changes.include，则只有被包含的事件类型才有资格；
然后应用 schema-changes.exclude；
当某类型同时出现在两个列表中时，exclude 优先。

source {
  MySQL-CDC {
    # ...
    schema-changes.enabled = true
    schema-changes.include = ["add.column", "drop.column"]
    schema-changes.exclude = ["change.column"]
  }
}

**排除 drop.column 时的数据处理方式。对于被保留的 NOT NULL 列，写入 NULL 会被 sink 拒绝，因此对一个源端已不再供数的 NOT NULL 列排除 drop.column 会在 sink 端失败。

表名支持正则以读取多个表

table-pattern 和 table-names 只能选择一个

env {
  # You can set engine configuration here
  parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 5000
  read_limit.bytes_per_second=7000000
  read_limit.rows_per_second=400
}

source {
  MySQL-CDC {
    server-id = 5652
    username = "st_user_source"
    password = "mysqlpw"
    database-pattern = "source.*"
    table-pattern = "source.*\\..*"
    url = "jdbc:mysql://mysql_cdc_e2e:3306"
  }
}

sink {
  Console {
  }
}

常见问题

MySQL CDC 需要哪些权限？

MySQL 用户需要以下权限：

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

同时需要在 my.cnf / my.ini 中开启 binlog：

[mysqld]
log_bin = mysql-bin
binlog_format = ROW
binlog_row_image = FULL

SeaTunnel 能从 MySQL 从库读取 CDC 数据吗？

可以。SeaTunnel 通过订阅 binlog 工作，从库也有 binlog 流。将 SeaTunnel 指向从库可以减轻主库压力。需确保从库开启了 binlog，并在配置中设置 log_slave_updates = ON。

MySQL CDC 是否支持无主键表？

默认要求主键。如果源表没有声明主键，但存在其他可唯一标识记录的列，可以像当前文档中的 source options 示例那样，通过 table-names-config.primaryKeys 指定自定义主键。若没有稳定的唯一键，下游就无法安全处理 UPDATE / DELETE 事件。

全量快照阶段如何工作？何时切换为增量读取？

首次启动时，SeaTunnel 对已配置的表做一次一致性全量快照。快照完成后，自动从快照开始时记录的 binlog 位置切换为增量读取，确保切换过程中不丢失任何变更事件。

MySQL CDC 是否支持 DDL 传播？

支持，但能力有限。需要启用 schema-changes.enabled = true，并遵循当前页面以及 Schema Evolution 文档中已经定义好的契约。目前文档中明确支持 add column、drop column、rename column 和 modify column。

运行多个 CDC 任务时如何避免 `server-id` 冲突？

每个 CDC 任务必须使用唯一的 server-id 或不重叠的范围。重复的 server-id 会导致 MySQL 服务器断开其中一个客户端连接。建议为每个任务分配独立的范围，例如一个任务用 5400-5600，另一个用 5601-5800。

初始快照为什么很慢？

快照速度取决于表大小、JDBC fetch size 和网络带宽。可以通过调整 snapshot.split.size 和 snapshot.fetch.size 来控制快照切分与抓取行为。对于不需要历史数据的大表，可将 startup.mode 设为 "latest"，从最新 offset 启动并跳过初始快照。

如何处理时区和字符集问题？

将 server-time-zone 设置为与 MySQL 服务器一致的时区，例如 "Asia/Shanghai"。字符集问题可通过在 JDBC 连接 URL 中追加 characterEncoding=UTF-8&useUnicode=true 来解决。

另请参阅

若需要一份面向生产的端到端实践指南，涵盖全量 + 增量同步生命周期、2PC sink 配置、Schema 演进与常见故障排查，请参阅 CDC 生产实战手册。