【Flink CDC（一）】实现mysql整表与增量读取

文章目录

一. 运行前准备
- 1. 依赖
- - 1.1. Maven dependency
  - 1.2. SQL Client JAR（推荐）
- 2. 配置 MySQL 服务器（必须）
二. 功能说明
- 1. 启动模式
- 2. 全量阶段支持 checkpoint
- 3. 关于无主键表
- Exactly-Once 处理
三. 实战
- 1. 实现mysql整表与增量表同步
FAQ

MySQL CDC 连接器允许从 MySQL 数据库读取快照数据（比如：flink任务消费时刻的整表数据）和增量数据。本文描述了如何设置 MySQL CDC 连接器来对 MySQL 数据库运行 SQL 查询。

本篇只关注mysql整表与增量读取的实现，对于并发读取等能力后续再探索。

一. 运行前准备

1. 依赖

1.1. Maven dependency

<dependency><groupId>com.ververica</groupId><artifactId>flink-connector-mysql-cdc</artifactId><!-- 请使用已发布的版本依赖，snapshot版本的依赖需要本地自行编译。 --><version>2.4.0</version>
</dependency>

1.2. SQL Client JAR（推荐）

下载 flink-sql-connector-mysql-cdc-2.4.0.jar 到 <FLINK_HOME>/lib/ 目录下。

2. 配置 MySQL 服务器（必须）

你必须定义一个 MySQL 用户，该用户对 MySQL CDC 连接器监视的所有数据库都应该具有所需的权限。

# 创建用户
mysql> CREATE USER 'user'@'localhost' IDENTIFIED BY 'password';# 赋权
mysql> GRANT SELECT, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'user' IDENTIFIED BY 'password';# 刷新权限
mysql> FLUSH PRIVILEGES;

注意：

在 scan.incremental.snapshot.enabled 参数已启用时（默认情况下已启用）时，不再需要授予 reload 权限。

二. 功能说明

1. 启动模式

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

initial （默认）：在第一次启动时对受监视的数据库表执行初始快照，并继续读取最新的 binlog。

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

latest-offset：首次启动时，从不对受监视的数据库表执行快照，连接器仅从 binlog
的结尾处开始读取，这意味着连接器只能读取在连接器启动之后的数据更改。

specific-offset：跳过快照阶段，从指定的 binlog 位点开始读取。位点可通过 binlog 文件名和位置指定，或者在 GTID 在集群上启用时通过 GTID 集合指定。

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

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()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.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.mode' = 'timestamp', -- 从特定位点启动'scan.startup.timestamp-millis' = '1667232000000' -- 在时间戳启动模式下指定启动时间戳...
)

2. 全量阶段支持 checkpoint

增量快照读取提供了在区块级别执行检查点的能力。它使用新的快照读取机制解决了以前版本中的检查点超时问题。

3. 关于无主键表

从2.4.0 版本开始支持无主键表，使用无主键表必须设置 scan.incremental.snapshot.chunk.key-column，且只能选择非空类型的一个字段。

在使用无主键表时，需要注意以下两种情况。

配置 scan.incremental.snapshot.chunk.key-column 时，如果表中存在索引，请尽量使用索引中的列来加快 select 速度。

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

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

Exactly-Once 处理

MySQL CDC 连接器是一个 Flink Source 连接器，它将首先读取表快照块，然后继续读取 binlog，无论是在快照阶段还是读取 binlog 阶段，MySQL CDC 连接器都会在处理时准确读取数据，即使任务出现了故障。

三. 实战

1. 实现mysql整表与增量表同步

-- 'scan.startup.mode'= 'initial' 
-- 
CREATE TABLE tjy_sql1  
(  `id` int,  `name` string,  `face` string  ,PRIMARY KEY(id) NOT ENFORCED  
) WITH (  'connector' = 'mysql-cdc',  'hostname' = 'xxx',  'port' = '3306',  'username' = 'middle_test',  'password' = '123456',  'database-name' = 'middle_test',  'table-name' = 'tjy_fortest1'  -- ,'scan.incremental.snapshot.enabled' = 'false'  --  initial: 默认值，全表同步，然后进行增量同步；--  'scan.startup.mode'= 'initial'  -- 'debezium.snapshot.mode' = 'initial'      );  CREATE TABLE tjy_sql1_sink  (  `id` int,  `name` string,  `face` string  ,PRIMARY KEY(id) NOT ENFORCED  ) WITH (  'connector' = 'mysql-x',  'url' = 'jdbc:mysql://xxx:3306/middle_test?useunicode=true&characterEncoding=utf8&useSSL=false&useCursorFetch=true',  'username' = 'middle_test',  'password' = '123456',  'table-name' = 'flink_type',  'table-name' = 'tjy_fortest2'  );  insert into tjy_sql1_sink select * from tjy_sql1;