MySQL的CDC源表

什么是MySQL的CDC源表

MySQL的CDC源表，即MySQL的流式源表，会先读取数据库的历史全量数据，并平滑切换到Binlog读取上，保证不多读一条也不少读一条数据。即使发生故障，也能保证通过Exactly Once语义处理数据。MySQL CDC Connector支持并发地读取全量数据，通过增量快照算法实现了全程无锁和断点续传。

实现原理

Source在启动时会扫描全表，将表按照主键分成多个chunk。并使用增量快照算法逐个读取每个chunk的数据。作业会周期性执行Checkpoint，记录下已经完成的chunk。当发生Failover时，只需要继续读取未完成的chunk。当chunk全部读取完后，会从之前获取的Binlog位点读取增量的变更记录。Flink作业会继续周期性执行Checkpoint，记录下Binlog位点，当作业发生Failover，便会从之前记录的Binlog位点继续处理，从而实现Exactly Once语义。更详细的增量快照算法，请参见MySQL CDC Connector。

前提条件

• MySQL和VVP的网络连通。
• MySQL服务器配置如下：
  • MySQL版本为5.7和8.0.X。
  • 已开启了Binlog。
  • Binlog格式已设置为ROW。
  • binlog_row_image已设置为FULL。
• 已在MySQL配置文件中配置了交互超时或等待超时参数。
• 已创建MySQL用户，并授予了SELECT、 SHOW DATABASES 、REPLICATION SLAVE和REPLICATION CLIENT权限。

使用限制

• 仅vvr-4.0.8-flink-1.13及以上引擎版本支持无锁读取和并发读取功能。
• MySQL CDC Connector支持读取的MySQL版本为5.7和8.0.X。
  说明 VVR 4.0.11及以上版本支持读取MySQL 5.6。
• MySQL CDC 源表暂不支持定义Watermark。如果您需要进行窗口聚合，您可以采用非窗口聚合的方式，详情请参见不支持定义Watermark，那如何进行窗口聚合？。
• MySQL的CDC源表需要一个有特定权限（包括SELECT、SHOW DATABASES、REPLICATION SLAVE和REPLICATION CLIENT）的MySQL用户，才能读取全量和增量数据。

注意事项

• 每个作业需显式配置不同的Server ID。

  每个同步数据库数据的客户端，都会有一个唯一ID，即Server ID。MySQL SERVER会根据该ID来维护网络连接以及Binlog位点。因此如果有大量不同的Server ID的客户端一起连接MySQL SERVER，可能导致MySQL SERVER的CPU陡增，影响线上业务稳定性。

  此外，多个作业共享相同的Server ID，会导致Binlog位点错乱，多读或少读数据。因此建议每个CDC作业都配置不同的Server ID。建议通过动态Hints来配置Server ID，而不是在DDL参数中配置Server ID。配置不同的Server ID代码示例如下。

  SELECT * FROM source_table  /*+ OPTIONS('server-id'='123456') */ ;

  动态Hints详情请参见动态Hints。

  说明 如果开启增量快照读取数据，则server-id配置项需要配置成与作业并发匹配的Server ID范围。
• 仅VVR 4.0.8 及以上版本支持全量阶段的无锁读取、并发读取、断点续传等功能。

  如果您使用的是VVR 4.0.8以下版本，需要对MySQL用户授予RELOAD权限用来获取全局读锁，保证数据读取的一致性。全局读锁会阻塞写入操作，持锁时间可能达到秒级，因此可能对线上业务造成影响。

  此外，VVR 4.0.8以下版本在全量读取阶段无法执行Checkpoint，全量阶段的作业失败会导致作业重新读取全量数据，稳定性不佳。因此建议您将作业升级到VVR 4.0.8及以上版本。

DDL定义

CREATE TABLE mysqlcdc_source (
   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' = '<yourHostname>',
  'port' = '3306',
  'username' = '<yourUsername>',
  'password' = '<yourPassword>',
  'database-name' = '<yourDatabaseName>',
  'table-name' = '<yourTableName>'
);

WITH参数

元数据

元数据在分库分表合并同步场景非常实用，因为分库分表合并后，一般业务还是希望区分每条数据的库名和表名来源，而元数据列可以访问源表的库名和表名信息。因此通过元数据列可以非常方便地将多张分表合并到一张目的表。

CREATE TABLE mysql_orders (
  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' = 'flinkuser',
  'password' = 'flinkpw',
  'database-name' = 'mydb_.*', -- 正则匹配多个分库。
  'table-name' = 'orders_.*'   -- 正则匹配多张分表。
);

INSERT INTO holo_orders SELECT * FROM mysql_orders;

并发控制

MySQL-CDC connector 支持多并发读取全量数据，能够提高数据加载效率。同时配合Flink VVP平台的Autopilot自动调优功能，在多并发读取完成后增量阶段，能够自动缩容，节约计算资源。

Autopilot自动缩容

全量阶段积累了大量历史数据，为了提高读取效率，通常采用并发的方式读取历史数据。而在Binlog增量阶段，因为Binlog数据量少且为了保证全局有序，通常只需要单并发读取。全量阶段和增量阶段对资源的不同需求，可以通过自动调优功能自动帮您实现性能和资源的平衡。

类型映射

常见问题

• 不支持定义Watermark，那如何进行窗口聚合？
• 如何跳过Snapshot阶段，只从变更数据开始读取？
• 如何读取一个分库分表的MySQL数据库？
• 全表读取阶段效率慢、存在反压，应该如何解决？
• 如何判断CDC作业是否已完成全量数据同步？
• 报错：com.github.shyiko.mysql.binlog.network.ServerException
• 报错：The primary key is necessary when enable 'Key: 'scan.incremental.snapshot.enabled'

技术原理及企业版特性

• Flink CDC企业版特性

• Flink CDC技术