实时计算Flink版：MySQL

背景信息

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

特色功能

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

前提条件

• CDC源表和数据摄入数据源

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

  • MySQL和VVP的网络连通。

  • MySQL服务器配置要求：

    • MySQL版本为5.6，5.7或8.0.x。

    • 已开启了Binlog。

    • Binlog格式已设置为ROW。

    • binlog_row_image已设置为FULL。

  • 已在MySQL配置文件中配置了交互超时或等待超时参数。

  • 已创建MySQL用户，并授予了SELECT、SHOW DATABASES、REPLICATION SLAVE和REPLICATION CLIENT权限。

• 维表和结果表

  • 已创建MySQL数据库和表，详情请参见RDS MySQL创建数据库和账号、PolarDB MySQL创建数据库和账号或自建MySQL创建数据库和账号。

  • 已设置IP白名单，详情请参见RDS MySQL白名单设置、PolarDB MySQL白名单设置或自建MySQL白名单设置。

使用限制

• CDC源表和数据摄入数据源

  • 仅VVR 4.0.8及以上引擎版本支持无锁读取和并发读取功能。

  • 请根据MySQL版本选择合适的引擎版本，MySQL版本支持情况如下表所示。

    您可以通过执行select version()命令来查看MySQL的版本。

    VVR版本	支持的MySQL版本
    VVR 4.0.8 ～ VVR 4.0.10	5.7
    	8.0.x
    VVR 4.0.11及以上版本	5.6.x
    	5.7.x
    	8.0.x

    重要

    为了确保RDS MySQL 5.6.x版本的正常运行，默认已开启增量快照功能（即scan.incremental.snapshot.enabled=true），且不支持关闭增量快照功能，而RDS MySQL 6.0.8和8.0.1版本的数据库已解除该限制，即支持关闭增量快照功能。建议您不要关闭增量快照功能，因为关闭增量快照功能会锁定MySQL数据库，可能会对线上业务处理性能产生影响。

  • MySQL CDC源表暂不支持定义Watermark。

  • MySQL的CDC源表需要一个有特定权限（包括SELECT、SHOW DATABASES、REPLICATION SLAVE和REPLICATION CLIENT）的MySQL用户，才能读取全量和增量数据。

  • 当结合CTAS和CDAS整库同步语法使用时，MySQL CDC源表可以同步部分Schema变更，支持的变更类型详情请参见表结构变更同步策略。在其他使用场景下，MySQL CDC源表无法同步Schema变更操作。

  • MySQL CDC源表无法同步Truncate操作。

  • 对于RDS MySQL，不建议通过备库或只读从库读取数据。因为RDS MySQL的备库和只读从库Binlog保留时间默认很短，可能由于Binlog过期清理，导致作业无法消费Binlog数据而报错。

  • MySQL CDC源表不支持读取PolarDB MySQL版1.0.19及以前版本的多主架构集群（什么是多主集群？）。PolarDB MySQL版1.0.19及以前版本的多主架构集群产生的Binlog可能出现重复Table id，导致CDC源表Schema映射错误，从而解析Binlog数据报错。PolarDB MySQL版在高于1.0.19的版本进行适配，保证Binlog内Table id不会出现重复，从而避免解析报错。

• 维表和结果表

  • Flink计算引擎VVR 4.0.11及以上版本支持MySQL连接器。

  • 语义上可以保证At-Least-Once，在结果表有主键的情况下，幂等可以保证数据的正确性。

注意事项

• CDC源表和数据摄入数据源

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

    • Server ID作用

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

      此外，如果多个MySQL CDC数据源共享相同的Server ID，且数据源之间无法复用时，会导致Binlog位点错乱，多读或少读数据。还可能出现Server ID冲突的报错，详情请参见上下游存储。因此建议每个MySQL CDC数据源都配置不同的Server ID。

    • Server ID配置方式

      Server ID可以在DDL中指定，也可以通过动态Hints配置。

      建议通过动态Hints来配置Server ID，而不是在DDL参数中配置Server ID。动态Hints详情请参见动态Hints。

    • 不同场景下Server ID的配置

      • 未开启增量快照框架或并行度为1

        当未开启增量快照框架或并行度为1时，可以指定一个特定的Server ID。

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

      • 开启增量快照框架且并行度大于1

        当开启增量快照框架且并行度大于1时，需要指定Server ID范围，要保证范围内可用的Server ID数量不小于并行度。假设并行度为3，可以如下配置：

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

      • 结合CTAS进行数据同步

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

      • 同一作业包含多个MySQL CDC源表（非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;

  • 仅VVR 4.0.8及以上版本支持全量阶段的无锁读取、并发读取、断点续传等功能。

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

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

• 结果表

  • RDS MySQL数据库支持自增主键，因此在结果表的DDL中不声明该自增字段。例如ID是自增字段，Flink DDL不声明该自增字段，则数据库在一行数据写入过程中会自动填补相关自增字段。

  • 结果表的DDL声明的字段必须至少存在一个非主键的字段，否则产生报错。

  • 结果表的DDL中NOT ENFORCED表示Flink自身对主键不做强制校验，需要您自行保证主键的正确性和完整性。

    Flink并不充分支持强制校验，Flink将假设列的可为空性与主键中的列是对齐的，从而认为主键是正确的，详情请参见Validity Check。

• 维表

  如果做维表时希望使用索引查询，请按照MySQL最左前缀原则排列JOIN指定的数据列。但这并无法保证使用索引，由于SQL优化，某些条件可能会被优化导致连接器得到的过滤条件无法命中索引。要确定连接器是否真正使用了索引进行查询，可以在数据库侧查看具体执行的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。

• 元数据

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

  自vvr-4.0.11-flink-1.13版本开始，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 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。

  重要

  table-name和database-name也支持使用逗号（,）分隔形式指定多张表或多个库，例如 'table-name' = 'mytable1,mytable2'。但是其与正则表达式中的逗号（,）冲突，所以如果使用了含有逗号的正则表达式，则正则表达式需要改写成竖线（|）的形式，例如mytable_\d{1, 2)需要改写成等价的正则表达式(mytable_\d{1}|mytable_\d{2})，来避免使用逗号。

• 并发控制

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

  在Flink全托管控制台，您可以在资源配置页面的基础模式或专家模式中设置作业的并发数。设置并发的区别如下：

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

  • 专家模式支持按需为某个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：仅对指定表的变更事件进行解析。

  • 使用配置项scan.only.deserialize.captured.tables.changelog.enabled：采用多线程对Binlog文件进行解析、并按顺序投放到消费队列。

• 优化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',  -- 仅对指定表的变更事件进行解析。
    'scan.parallel-deserialize-changelog.enabled' = 'true'  -- 使用多线程对Binlog进行解析。
    ...
)

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
  scan.parallel-deserialize-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问题。