MySQL

背景信息

MySQL连接器支持所有兼容MySQL协议的数据库，包括RDS MySQL、PolarDB for MySQL、OceanBase（MySQL模式）或者自建MySQL。

特色功能

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

前提条件

在使用MySQL CDC源表前，必须先按照配置MySQL进行操作，这些操作主要为了满足使用MySQL CDC源表的前提条件

使用限制

注意事项

• 每个MySQL CDC数据源需显式配置不同的Server ID

  Server ID作用

  每个MySQL CDC数据源需显式配置不同的Server ID。如果多个MySQL CDC数据源共享相同的Server ID，且数据源之间无法复用时，会导致Binlog位点错乱，数据多读或少读。

  不同场景下Server ID的配置

  Server ID可以在DDL中指定，但建议通过动态Hints来配置Server ID，而不是在DDL参数中配置Server ID。

  • 并行度 = 1 或未开增量快照

    ## 当未开启增量快照框架或并行度为1时，可以指定一个特定的Server ID。
    SELECT * FROM source_table /*+ OPTIONS('server-id'='123456') */ ;

  • 并行度 > 1 且开增量快照

    ## 需要指定Server ID范围，要保证范围内可用的Server ID数量不小于并行度。假设并行度为3
    SELECT * FROM source_table /*+ OPTIONS('server-id'='123456-123458') */ ;

  • CTAS进行数据同步

    当结合CTAS进行数据同步时，如果CDC数据源配置相同，会自动对数据源进行复用，此时可以为多个CDC数据源配置相同的Server ID。详情请参见代码示例四：多CTAS语句。

  • 非 CTAS 多源表且不可复用

    当作业中包含多个MySQL CDC源表，且不是使用CTAS语句同步时，数据源无法进行复用，需要为每一个CDC源表提供不同的Server ID。同理，如果开启增量快照框架且并行度大于1，需要指定Server ID范围。

    select * from 
      source_table1 /*+ OPTIONS('server-id'='123456-123457') */
    left join 
      source_table2 /*+ OPTIONS('server-id'='123458-123459') */
    on source_table1.id=source_table2.id;

• 结果表

  • 自增主键，在DDL中不声明。写入数据时，MySQL会自动填写。

  • 必须至少声明一个非主键字段，否则报错。

  • DDL中NOT ENFORCED表示Flink自身对主键不做强制校验，需要您自行保证主键的正确性和完整性。详情请参见Validity Check。

• 维表

  如果希望使用索引加速查询，JOIN时字段顺序，要和索引定义的顺序一致（最左前缀原则）。比如索引是 (a, b, c)，JOIN条件则为ON t.a = x AND t.b = y。

  Flink 生成的 SQL 可能被优化器改写，导致实际查库时无法命中索引。确认是否使用了索引，去 MySQL 里看执行计划（EXPLAIN）或慢查询日志，查看真实执行的Select语句。

使用示例

• CDC源表

  CREATE TEMPORARY 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',
    'hostname' = '<yourHostname>',
    'port' = '3306',
    'username' = '<yourUsername>',
    'password' = '<yourPassword>',
    'database-name' = '<yourDatabaseName>',
    'table-name' = '<yourTableName>'
  );
  
  CREATE TEMPORARY TABLE blackhole_sink(
    order_id INT,
    customer_name STRING
  ) WITH (
    'connector' = 'blackhole'
  );
  
  INSERT INTO blackhole_sink
  SELECT order_id, customer_name FROM mysqlcdc_source;

• 维表

  CREATE TEMPORARY TABLE datagen_source(
    a INT,
    b BIGINT,
    c STRING,
    `proctime` AS PROCTIME()
  ) WITH (
    'connector' = 'datagen'
  );
  
  CREATE TEMPORARY TABLE mysql_dim (
    a INT,
    b VARCHAR,
    c VARCHAR
  ) WITH (
    'connector' = 'mysql',
    'hostname' = '<yourHostname>',
    'port' = '3306',
    'username' = '<yourUsername>',
    'password' = '<yourPassword>',
    'database-name' = '<yourDatabaseName>',
    'table-name' = '<yourTableName>'
  );
  
  CREATE TEMPORARY TABLE blackhole_sink(
    a INT,
    b STRING
  ) WITH (
    'connector' = 'blackhole'
  );
  
  INSERT INTO blackhole_sink
  SELECT T.a, H.b
  FROM datagen_source AS T JOIN mysql_dim FOR SYSTEM_TIME AS OF T.`proctime` AS H ON T.a = H.a;

• 结果表

  CREATE TEMPORARY TABLE datagen_source (
    `name` VARCHAR,
    `age` INT
  ) WITH (
    'connector' = 'datagen'
  );
  
  CREATE TEMPORARY TABLE mysql_sink (
    `name` VARCHAR,
    `age` INT
  ) WITH (
    'connector' = 'mysql',
    'hostname' = '<yourHostname>',
    'port' = '3306',
    'username' = '<yourUsername>',
    'password' = '<yourPassword>',
    'database-name' = '<yourDatabaseName>',
    'table-name' = '<yourTableName>'
  );
  
  INSERT INTO mysql_sink
  SELECT * FROM datagen_source;

• 数据摄入数据源

  source:
    type: mysql
    name: MySQL Source
    hostname: ${mysql.hostname}
    port: ${mysql.port}
    username: ${mysql.username}
    password: ${mysql.password}
    tables: ${mysql.source.table}
    server-id: 7601-7604
  
  sink:
    type: values
    name: Values Sink
    print.enabled: true
    sink.print.logger: true

关于MySQL CDC源表

• 实现原理

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

  更详细的增量快照算法，请参见MySQL CDC Connector。

• 元数据

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

  MySQL CDC Source支持元数据列语法，您可以通过元数据列访问以下元数据。

  元数据key	元数据类型	描述
  database_name	STRING NOT NULL	包含该行记录的库名。
  table_name	STRING NOT NULL	包含该行记录的表名。
  op_ts	TIMESTAMP_LTZ(3) NOT NULL	该行记录在数据库中的变更时间，如果该记录来自表的存量历史数据而不是Binlog中获取，则该值总是0。
  op_type	STRING NOT NULL	该行记录的变更类型。 +I：表示INSERT消息 -D：表示DELETE消息 -U：表示UPDATE_BEFORE消息 +U：表示UPDATE_AFTER消息 说明 仅实时计算引擎VVR 8.0.7及以上版本支持。

  将MySQL实例中多个分库下的多张orders表，合并同步到下游Hologres的holo_orders表中，代码示例如下所示。

  CREATE TEMPORARY 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, -- 读取变更时间。
    op_type STRING METADATA FROM 'op_type' 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;

  在上面代码的基础上，WITH参数里配置scan.read-changelog-as-append-only.enabled参数为true时，输出结果根据下游表主键设置情况有不同的表现：

  • 下游表主键为order_id时，输出结果仅包含上游表每个主键的最后一次变更。即对于某个主键最后一次变更为删除操作的数据，在下游表可以看到一条相同主键的、op_type为-D的数据。

  • 下游表主键为order_id、operation_ts、op_type时，输出结果包含上游表每个主键的完整变更。

• 支持正则表达式

  MySQL CDC源表支持在表名或者库名中使用正则表达式匹配多个表或者多个库。通过正则表达式指定多张表的代码示例如下。

  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)' -- 正则表达式匹配多张表。
  );

  上述例子中的正则表达式解释：

  • ^(test).* 是前缀匹配示例，这个表达式可以匹配以test开头的库名，例如test1或test2。

  • .*[p$] 是后缀匹配示例， 这个表达式可以匹配以p结尾的库名，例如cdcp或edcp。

  • txc是指定匹配， 可以匹配指定名称的数据库名，例如txc。

  MySQL CDC在匹配全路径表名时，会通过库名和表名来唯一确定一张表，即使用database-name.table-name作为匹配表的模式，例如匹配模式 (^(test).*|^(tpc).*|txc|.*[p$]|t{2}).(t[ 5-8]|tt) 就可以匹配到数据库中的表txc.tt和test2.test5。

  重要

  在SQL作业的配置中，table-name和database-name不支持使用逗号（,）分隔形式指定多张表或多个库。

  • 如果需要匹配多个表或使用多个正则表达式，可以用竖线（|）连接并用小括号包围，例如需要读取表user和product，table-name可以配置为(user|product)。

  • 如果正则表达式包含逗号，需要用竖线（|）运算符进行改写，例如正则表达式mytable_\d{1, 2}需要改写成等价的(mytable_\d{1}|mytable_\d{2})，来避免使用逗号。

• 并发控制

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

  在实时计算开发控制台，您可以在资源配置页面的基础模式或专家模式中设置作业的并发数。设置并发的区别如下：

  • 基础模式设置的并发数为整个作业的全局并发数。

  • 专家模式支持按需为某个VERTEX设置并发数。

  资源配置详情请参见配置作业部署信息。

  重要

  无论是基础模式还是专家模式，在设置并发时，表中声明的server-id范围必须大于等于作业的并发数。例如server-id的范围为5404-5412，则共有8个唯一的server-id，因此作业最多可以设置8个并发，且不同的作业对于同一个MySQL实例的server-id范围不能有重叠，即每个作业需显式配置不同的server-id。

• Autopilot自动缩容

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

  自动调优会监控MySQL CDC Source的每个task的流量。当进入Binlog阶段，如果只有一个task在负责Binlog读取，其他task均空闲时，自动调优便会自动缩小Source的CU数和并发。开启自动调优只需要在作业运维页面，将自动调优的模式设置为Active模式。

  说明

  默认调低并发度的最小触发时间间隔为24小时。更多自动调优的参数和细节，请参见配置自动调优。

• 启动模式

  使用配置项scan.startup.mode可以指定MySQL CDC源表的启动模式。可选值包括：

  • initial （默认）：在第一次启动时对数据库表进行全量读取，完成后切换至增量模式读取Binlog。

  • earliest-offset：跳过快照阶段，从可读取的最早Binlog位点开始读取。

  • latest-offset：跳过快照阶段，从Binlog的结尾处开始读取。该模式下源表只能读取在作业启动之后的数据变更。

  • specific-offset：跳过快照阶段，从指定的Binlog位点开始读取。位点可通过Binlog文件名和位置指定，或者使用GTID集合指定。

  • timestamp：跳过快照阶段，从指定的时间戳开始读取Binlog事件。

  使用示例：

  CREATE TABLE mysql_source (...) WITH (
      'connector' = 'mysql-cdc',
      'scan.startup.mode' = 'earliest-offset', -- 从最早位点启动。
      'scan.startup.mode' = 'latest-offset', -- 从最晚位点启动。
      'scan.startup.mode' = 'specific-offset', -- 从特定位点启动。
      'scan.startup.mode' = 'timestamp', -- 从特定位点启动。
      'scan.startup.specific-offset.file' = 'mysql-bin.000003', -- 在特定位点启动模式下指定Binlog文件名。
      'scan.startup.specific-offset.pos' = '4', -- 在特定位点启动模式下指定Binlog位置。
      'scan.startup.specific-offset.gtid-set' = '24DA167-0C0C-11E8-8442-00059A3C7B00:1-19', -- 在特定位点启动模式下指定GTID集合。
      'scan.startup.timestamp-millis' = '1667232000000' -- 在时间戳启动模式下指定启动时间戳。
      ...
  )

  重要

  • MySQL source会在Checkpoint时将当前位点以INFO级别打印到日志中，日志前缀为Binlog offset on checkpoint {checkpoint-id}，该日志可以帮助您将作业从某个Checkpoint位点开始启动作业。

  • 如果读取的表曾经发生过表结构变化，从最早位点（earliest-offset）、特定位点（specific-offset）或时间戳（timestamp）启动可能会发生错误。因为Debezium读取器会在内部保存当前的最新表结构，结构不匹配的早期数据无法被正确解析。

• 关于无主键CDC源表

  • 在Flink计算引擎VVR 6.0.7及以上版本支持使用MySQL CDC无主键源表，使用无主键表要求必须设置scan.incremental.snapshot.chunk.key-column，且只能选择非空类型的字段。

  • 无主键CDC源表的处理语义由scan.incremental.snapshot.chunk.key-column指定的列的行为决定：

    • 如果指定的列不存在更新操作，此时可以保证Exactly once语义。

    • 如果指定的列发生更新操作，此时只能保证At least once语义。但可以结合下游，通过指定下游主键，结合幂等性操作来保证数据的正确性。

• 读取阿里云RDS MySQL备份日志

  MySQL CDC源表支持读取阿里云RDS MySQL的备份日志。这在全量阶段执行时间较长，本地Binlog文件已经被自动清理，而自动或者手动上传的备份文件依然存在的场景下非常适用。

  使用示例：

  CREATE TABLE mysql_source (...) WITH (
      'connector' = 'mysql-cdc',
      'rds.region-id' = 'cn-beijing',
      'rds.access-key-id' = 'xxxxxxxxx', 
      'rds.access-key-secret' = 'xxxxxxxxx', 
      'rds.db-instance-id' = 'rm-xxxxxxxxxxxxxxxxx', 
      'rds.main-db-id' = '12345678',
      'rds.download.timeout' = '60s'
      ...
  )

• 开启CDC Source复用

  当同一个作业中有多个MySQL CDC源表时，每个源表都会启动对应的Binlog Client，如果源表数量较多并且读取的MySQL表都在同一个实例中时，会对数据库造成较大压力，详情请参见MySQL CDC常见问题。

  实时计算引擎VVR 8.0.7及以上版本支持MySQL CDC Source复用，当不同的CDC源表配置项除了数据库、表名和server-id外的其他配置项均相同时，可以进行合并。开启Source复用后，实时计算引擎会尽可能将同一个作业中能够合并的MySQL CDC源表进行合并。

  您可以在SQL作业中使用SET命令开启source复用功能：

  SET 'table.optimizer.source-merge.enabled' = 'true';

  对已有作业启用 Source 复用后，需要无状态启动。原因是 Source 复用会导致作业拓扑改变，从原有作业状态可能无法启动或者丢失数据。

  重要

  • VVR 8.0.8及8.0.9版本，在开启CDC Source复用时，还需要额外设置SET 'sql-gateway.exec-plan.enabled' = 'false'。

  • 在开启CDC Source复用后，不建议将作业配置项pipeline.operator-chaining设为false，因为将算子链断开后，Source发送给下游算子的数据会增加序列化和反序列的开销，当合并的Source越多时，开销会越大。

  • 在实时计算引擎VVR 8.0.7版本，将pipeline.operator-chaining设为false时会出现序列化的问题。

加速Binlog读取

MySQL连接器作为源表或数据摄入数据源使用时，在增量阶段会解析Binlog文件生成各种变更消息，Binlog文件使用二进制记录着所有表的变更，可以通过以下方式加速Binlog文件解析。

• 开启解析过滤配置

  • 使用配置项scan.only.deserialize.captured.tables.changelog.enabled：仅对指定表的变更事件进行解析。

• 优化Debezium参数

  debezium.max.queue.size: 162580
  debezium.max.batch.size: 40960
  debezium.poll.interval.ms: 50

  • debezium.max.queue.size：阻塞队列可以容纳的记录的最大数量。当Debezium从数据库读取事件流时，它会在将事件写入下游之前将它们放入阻塞队列。默认值为8192。

  • debezium.max.batch.size：该连接器每次迭代处理的事件条数最大值。默认值为2048。

  • debezium.poll.interval.ms：连接器应该在请求新的变更事件前等待多少毫秒。默认值为1000毫秒，即1秒。

使用示例：

CREATE TABLE mysql_source (...) WITH (
    'connector' = 'mysql-cdc',
    -- Debezium配置
    'debezium.max.queue.size' = '162580',
    'debezium.max.batch.size' = '40960',
    'debezium.poll.interval.ms' = '50',
    -- 开启解析过滤
    'scan.only.deserialize.captured.tables.changelog.enabled' = 'true',  -- 仅对指定表的变更事件进行解析。
    ...
)

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: ${mysql.source.table}
  server-id: 7601-7604
  # Debezium配置
  debezium.max.queue.size: 162580
  debezium.max.batch.size: 40960
  debezium.poll.interval.ms: 50
  # 开启解析过滤
  scan.only.deserialize.captured.tables.changelog.enabled: true

MySQL CDC 企业版本binlog消费能力为85MB/s，约为开源社区的2倍，当Binlog文件产生速度大于 85MB/s 时（即每6s一个512MB大小的文件），Flink 作业的延迟会持续上升，在Binlog文件产生速度降低后处理延迟会逐步下降。在Binlog文件包含大事务时，可能会导致处理延迟短暂上升，读取完该事务的日志后处理延迟会下降。

MySQL CDC DataStream API

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
public class MySqlSourceExample {
  public static void main(String[] args) throws Exception {
    MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
        .hostname("yourHostname")
        .port(yourPort)
        .databaseList("yourDatabaseName") // set captured database
        .tableList("yourDatabaseName.yourTableName") // set captured table
        .username("yourUsername")
        .password("yourPassword")
        .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
        .build();
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // enable checkpoint
    env.enableCheckpointing(3000);
    env
      .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
      // set 4 parallel source tasks
      .setParallelism(4)
      .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering
    env.execute("Print MySQL Snapshot + Binlog");
  }
}

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-core</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-base</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-common</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-api-java-bridge</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>ververica-connector-mysql</artifactId>
    <version>${vvr.version}</version>
</dependency>

常见问题

CDC源表使用中可能遇到的问题，详情请参见CDC问题。

Flink CDC技术原理及企业版特性

• Flink CDC企业版特性

• Flink CDC技术