MySQL CDC 连接器提供了从 MySQL 数据库读取全量和增量数据的能力，仅用于做数据源表。

使用限制

• MySQL CDC 连接器暂时仅支持在 Flink 1.16-volcano 及以上引擎版本中使用。
• 支持 MySQL 版本为 5.6, 5.7, 8.x。
• 如果您需要使用 MySQL CDC 连接器连接云数据库 veDB MySQL 版，您的连接终端请按照以下要求配置，否则可能会因为自定义连接终端的限制而出现任务故障。
  如需详细了解各参数含义，请参见编辑连接终端。
  • 读写模式：配置为读写。
  • 一致性级别：配置为最终一致性。
  • 主节点接受读：关闭该选项。
  • 事务拆分：打开该选项。
• 基于合规要求，如果需要使用 MySQL CDC 相关功能，需要用户自行上传 MySQL Driver 。请前往 MySQL 官网下载 8.0.27 版本，并且上传到作业开发依赖文件中。具体操作可以参考 使用 JDBC 或者 MySQL-CDC 数据源 。

DDL 定义

CREATE TABLE orders (
    order_id bigint,
    order_product_id bigint,
    order_customer_id bigint,
    order_status varchar,
    order_update_time timestamp,
    PRIMARY KEY (order_id) NOT ENFORCED  -- 如果要同步的数据库表定义了主键, 则这里也需要定义主键。
     ) WITH (
     'connector' = 'mysql-cdc',
     'hostname' = 'localhost',
     'port' = '3306',
     'username' = 'flinkuser',
     'password' = 'flinkpw',
     'database-name' = 'mydb',
     'table-name' = 'orders'
     );

WITH 参数

支持的元数据

下表中的元数据可以在 DDL 中作为只读（虚拟）meta 列声明。

下述创建表示例展示元数据列的用法：

CREATE TABLE products (
    db_name STRING METADATA FROM 'database_name' VIRTUAL,
    table_name STRING METADATA  FROM 'table_name' VIRTUAL,
    operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
    order_id INT,
    order_date TIMESTAMP(0),
    customer_name STRING,
    price DECIMAL(10, 5),
    product_id INT,
    order_status BOOLEAN,
    PRIMARY KEY(order_id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'username' = 'root',
    'password' = '123456',
    'database-name' = 'mydb',
    'table-name' = 'orders'
);

最佳实践

分库分表同步

在多库分表场景下，通过正则表达式匹配多张库表，来实现合并多张分表到一张表的功能，示例如下：

CREATE TABLE products (
  db_name STRING METADATA FROM 'database_name' VIRTUAL,
  table_name STRING METADATA  FROM 'table_name' VIRTUAL,
  operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
  order_id INT,
  order_date TIMESTAMP(0),
  customer_name STRING,
  price DECIMAL(10, 5),
  product_id INT,
  order_status BOOLEAN,
  PRIMARY KEY(order_id) NOT ENFORCED
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = 'localhost',
  'port' = '3306',
  'username' = 'root',
  'password' = '123456',
  'database-name' = '(^(test).*|^(tpc).*|txc|.*[p$]|t{2})', -- 正则表达式匹配多个库。
  'table-name' = '(t[5-8]|tt)' -- 正则表达式匹配多张表。
);

进行库表匹配时，会使用正则表达式 database-name\\.table-name 来与MySQL表的全限定名做匹配，所以该例子使用 (^(test).*|^(tpc).*|txc|.*[p$]|t{2})\\.(t[5-8]|tt)，可以匹配到表 txc.tt、test2.test5。

全量拉取分片键指定

在全量读取阶段，对于数据库表比较大的情况下，可能会对数据库造成压力，MySQL CDC 在 2.0 版本后在全量同步阶段基于论文 DBLog Paper，MySql CDC Connector 实现了一种新的全量数据拉取方式，这种方式可以并行地拉取全量数据，并且不需要全局的读锁（FLUSH TABLES WITH READ LOCK）。这种算法会把 MySQL 的全量数据划分成不同的小的 chunk，并且可以并发的读取（详细原理可以参考文档）。用户可以配置用来划分 chunk 的 key 和 chunk size 来控制 chunk 的划分和每个 chunk 的大小。其中 chunk key 满足以下条件最优：

• 数字类型：方便根据每个 chunk 的大小，快速的计算每个 chunk 的 start 和 end 值；
• 分布均匀：保证落在每个 chunk 里的数据分布是均匀的，否则会造成大部分数据分布在少量的 chunk 内，在数据拉取阶段产生数据倾斜

FAQ

过滤 delete 消息

在业务使用过程中，经常有上游 MySQL Source 为了节省存储空间，定期把历史数据删除，只保留七天内的数据。但是业务需要 Sink 的 OLAP 系统保留历史全量的数据，因此需要把 Source 定期发送的 delete 消息都忽略掉。
解决方法：添加参数 'debezium.skipped.operations' = 'd'，则表示过滤 delete 消息。

丢状态重启

如果修改了作业的拓扑结构，比如增减了 Source 的字段，则不能从历史状态恢复，此时可以丢状态，并从 specific-offset 模式全新启动。步骤如下：

1. 获取当前位点信息，具体是打开 TM 日志，每次 checkpoint，日志里都会打印每个表的位点信息，包括 binlog file 和 binlog position。

2024-09-27 18:16:10,488 INFO  com.ververica.cdc.connectors.mysql.source.reader.MySqlSourceReader [] - Binlog offset for tables [test_db.test] on checkpoint 4: {transaction_id=null, ts_sec=0, file=mysql-binlog.000666, pos=147223270, kind=SPECIFIC, gtids=e6e6e8e3-6351-11ef-8340-52540bd1ce99:81433397-81559559, row=0, event=20, server_id=1}

2. 修改 SQL 里 MySQL CDC Source 的启动模式，从上述位点恢复。

create table
  mysql_cdc_source (
    `id` INT NOT NULL,
    `a` DECIMAL(10, 2) NOT NULL,
    `b` DOUBLE,
    `c` DATE,
    `d` VARCHAR,
    PRIMARY KEY (`id`) NOT ENFORCED
  )
WITH
  (
    'connector' = 'mysql-cdc',
    'hostname' = 'vedbm-xxx.pri.mysql.vedb.ivolces.com',
    'port' = '3306',
    'username' = 'xroot',
    'password' = 'xxxx',
    'database-name' = 'test_db',
    'table-name' = 'test',
    'scan.startup.mode' = 'specific-offset', -- 从指定的 binlog 位点开始读取。位点可以通过 binlog 文件名和位置指定
    'scan.startup.specific-offset.file' = 'binlog.000666',
    'scan.startup.specific-offset.pos' = '147223270'
  );

3. 重启作业，从全新状态启动。