MySQL-CDC

概述

MySQL Extract 节点允许从 MySQL 数据库中读取快照数据和增量数据。本文档介绍如何设置 MySQL Extract 节点以对 MySQL 数据库运行 SQL 查询。

支持的版本

依赖

为了设置 MySQL Extract 节点，下表提供了使用构建自动化工具（例如 Maven 或 SBT）和带有 Sort Connectors JAR 包的 SQL 客户端的两个项目的依赖关系信息。

Maven 依赖

<dependency>
    <groupId>org.apache.inlong</groupId>
    <artifactId>sort-connector-mysql-cdc</artifactId>
    <version>1.8.0</version>
</dependency>

连接 MySQL 数据库还需要 MySQL 驱动程序依赖项。请下载mysql-connector-java-8.0.21.jar 并将其放入 FLINK_HOME/lib/。

设置 MySQL 服务器

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

• 创建 MySQL 用户:

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

• 向用户授予所需的权限:

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

注意： 启用 scan.incremental.snapshot.enabled 时不再需要 RELOAD 权限（默认启用）。

• 刷新用户的权限：

mysql> FLUSH PRIVILEGES;

查看更多关于权限说明。

注意事项

为每个 Reader 设置一个不同的 SERVER ID

每一个读取 Binlog 的 MySQL 数据库客户端都应该有一个唯一的 Id，称为 Server Id。 MySQL 服务器将使用此 Id 来维护网络连接和 Binlog 位置。因此，如果不同的作业共享相同的服务器 Id，可能会导致从错误的 Binlog 位置读取。 因此，建议通过 [SQL Hints](https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/sql/hints. html), 例如假设源并行度为 4，那么我们可以使用 SELECT * FROM source_table /*+ OPTIONS('server-id'='5401-5404') */ ; 为 4 个 Source Reader 中的每一个分配唯一的服务器 Id。

设置 MySQL 会话超时

当为大型数据库制作初始一致快照时，您建立的连接可能会在读取表时超时。您可以通过在 MySQL 配置文件中配置 interactive_timeout 和 wait_timeout 来防止这种行为。

• interactive_timeout：服务器在关闭交互式连接之前等待其活动的秒数。请参阅 MySQL 文档。
• wait_timeout：服务器在关闭非交互式连接之前等待其活动的秒数。请参阅 MySQL 文档。

如何创建一个 MySQL Extract 节点

SQL API 用法

下面这个例子展示了如何用 Flink SQL 创建一个 MySQL Extract 节点:

-- 设置 Checkpoint 为 3000 毫秒                       
Flink SQL> SET 'execution.checkpointing.interval' = '3s';   
Flink SQL> CREATE TABLE mysql_extract_node (
     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-inlong',
     'hostname' = 'YourHostname',
     'port' = '3306',
     'username' = 'YourUsername',
     'password' = 'YourPassword',
     'database-name' = 'YourDatabaseName',
     'table-name' = 'YourTableName');
Flink SQL> SELECT * FROM mysql_extract_node;

InLong Dashboard 用法

• Choose the BINLOG Data Source

• Configure the MySQL Source

InLong Manager Client 用法

TODO: 将在未来支持此功能。

MySQL Extract 节点参数

可用的元数据字段

以下格式元数据可以作为表定义中的只读 (VIRTUAL) 列公开。

扩展的 CREATE TABLE 示例演示了使用这些元数据字段的语法：

CREATE TABLE `mysql_extract_node` (
      `id` INT,
      `name` STRING,
      `database_name` string METADATA FROM 'meta.database_name',
      `table_name`    string METADATA FROM 'meta.table_name',
      `op_ts`         timestamp(3) METADATA FROM 'meta.op_ts',
      `op_type` string METADATA FROM 'meta.op_type',
      `batch_id` bigint METADATA FROM 'meta.batch_id',
      `is_ddl` boolean METADATA FROM 'meta.is_ddl',
      `update_before` ARRAY<MAP<STRING, STRING>> METADATA FROM 'meta.update_before',
      `mysql_type` MAP<STRING, STRING> METADATA FROM 'meta.mysql_type',
      `pk_names` ARRAY<STRING> METADATA FROM 'meta.pk_names',
      `data` STRING METADATA FROM 'meta.data_canal',
      `sql_type` MAP<STRING, INT> METADATA FROM 'meta.sql_type',
      `ingestion_ts` TIMESTAMP(3) METADATA FROM 'meta.ts',
      PRIMARY KEY (`id`) NOT ENFORCED 
) WITH (
      'connector' = 'mysql-cdc-inlong', 
      'hostname' = 'YourHostname',
      'migrate-all' = 'true',
      'port' = '3306',                
      'username' = 'YourUsername',
      'password' = 'YourPassword',
      'database-name' = 'YourDatabase',
      'table-name' = 'YourTable',
      'row-kinds-filtered' = '+I'
 );

数据类型映射

特性

多库多表同步

Mysql Extract 节点支持整库、多表同步。开启该功能后，Mysql Extract 节点会将表的物理字段压缩成 ‘canal-json’ 格式的特殊元字段 ‘data_canal’，也可配置为 ‘debezium-json’ 格式的元数据字段 ‘data_debezium’。

配置参数：

CREATE TABLE 示例演示该功能语法：

CREATE TABLE `table_1`(
`data` STRING METADATA FROM 'meta.data_canal' VIRTUAL)
WITH (
'inlong.metric.labels' = 'groupId=1&streamId=1&nodeId=1',
'migrate-all' = 'true',
'connector' = 'mysql-cdc-inlong',
'hostname' = 'localhost',
'database-name' = 'test,test01',
'username' = 'root',
'password' = 'inlong',
'table-name' = 'test01\.a{2}[0-9]$, test\.[\s\S]*'
)

启动模式

配置选项scan.startup.mode指定 MySQL CDC 使用者的启动模式。有效枚举包括：

• initial （默认）：在第一次启动时对受监视的数据库表执行初始快照，并继续读取最新的 binlog。
• earliest-offset：跳过快照阶段，从可读取的最早 binlog 位点开始读取
• latest-offset：首次启动时，从不对受监视的数据库表执行快照， 连接器仅从 binlog 的结尾处开始读取，这意味着连接器只能读取在连接器启动之后的数据更改。
• specific-offset：跳过快照阶段，从指定的 binlog 位点开始读取。位点可通过 binlog 文件名和位置指定，或者在 GTID 在集群上启用时通过 GTID 集合指定。
• timestamp：跳过快照阶段，从指定的时间戳开始读取 binlog 事件。

例如使用 DataStream API:

MySQLSource.builder()
    .startupOptions(StartupOptions.earliest()) // 从最早位点启动
    .startupOptions(StartupOptions.latest()) // 从最晚位点启动
    .startupOptions(StartupOptions.specificOffset("mysql-bin.000003", 4L) // 从指定 binlog 文件名和位置启动
    .startupOptions(StartupOptions.specificOffset("24DA167-0C0C-11E8-8442-00059A3C7B00:1-19")) // 从 GTID 集合启动
    .startupOptions(StartupOptions.timestamp(1667232000000L) // 从时间戳启动
    ...
    .build()

使用 SQL:

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 的位点开始启动的场景。
• 如果捕获变更的表曾经发生过表结构变化，从最早位点、特定位点或时间戳启动可能会发生错误，因为 Debezium 读取器会在内部保存当前的最新表结构，结构不匹配的早期数据无法被正确解析。