MySQL - 实时计算Flink版 - 阿里云

背景信息

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

类别	详情
支持类型	源表、维表和结果表
运行模式	仅支持流模式
数据格式	暂不适用
特有监控指标	源表 currentFetchEventTimeLag：数据产生到拉取到Source Operator的间隔。该指标仅在Binlog阶段有效，Snapshot阶段该值恒为0。 currentEmitEventTimeLag：数据产生到离开Source Operator的间隔。该指标仅在Binlog阶段有效，Snapshot阶段该值恒为0。 sourceIdleTime：源表至今有多久不产生新数据。维表和结果表：无。
API种类	Datastream和SQL
是否支持更新或删除结果表数据	是

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

- - - ```
    SELECT * FROM source_table /*+ OPTIONS('server-id'='123456') */ ;
```
- ```
SELECT * FROM source_table /*+ OPTIONS('server-id'='123456-123458') */ ;
```
    - ```
    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;
```

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

参数	说明	是否必填	数据类型	默认值	备注
connector	表类型。	是	STRING	无	作为源表时，可以填写为 `mysql-cdc` 或者 `mysql` ，二者等价。作为维表或结果表时，固定值为 `mysql` 。
hostname	MySQL数据库的IP地址或者Hostname。	是	STRING	无	建议填写专有网络VPC地址。
username	MySQL数据库服务的用户名。	是	STRING	无	无。
password	MySQL数据库服务的密码。	是	STRING	无	无。
database-name	MySQL数据库名称。	是	STRING	无	作为源表时，数据库名称支持正则表达式以读取多个数据库的数据。使用正则表达式时，尽量不要使用 ^ 和 $ 符号匹配开头和结尾。具体原因详见 table-name 备注的说明。
table-name	MySQL表名。	是	STRING	无	作为源表时，表名支持正则表达式以读取多个表的数据。在读取多个MySQL表时，将多个CTAS语句作为一个作业提交，可以避免启用多个Binlog监听，提高性能和效率。详情请参见示例四：多个CTAS语句作为一个作业提交。使用正则表达式时，尽量不要使用 ^ 和 $ 符号匹配开头和结尾。具体原因详见以下说明。
port	MySQL数据库服务的端口号。	否	INTEGER	3306	无。

参数	说明	是否必填	数据类型	默认值	备注
server-id	数据库客户端的一个数字ID。	否	STRING	默认会随机生成一个5400~6400的值。	该ID必须是MySQL集群中全局唯一的。建议针对同一个数据库的每个作业都设置一个不同的ID。该参数也支持ID范围的格式，例如5400-5408。在开启增量读取模式时支持多并发读取，此时推荐设定为ID范围，使得每个并发使用不同的ID。
scan.incremental.snapshot.enabled	是否开启增量快照。	否	BOOLEAN	true	默认开启增量快照。增量快照是一种读取全量数据快照的新机制。与旧的快照读取相比，增量快照有很多优点，包括：读取全量数据时，Source可以是并行读取。读取全量数据时，Source支持chunk粒度的检查点。读取全量数据时，Source不需要获取全局读锁（FLUSH TABLES WITH read lock）。如果您希望Source支持并发读取，每个并发的Reader需要有一个唯一的服务器ID，因此server-id必须是5400-6400这样的范围，并且范围必须大于等于并发数。
scan.incremental.snapshot.chunk.size	每个chunk的大小（包含的行数）。	否	INTEGER	8096	当开启增量快照读取时，表会被切分成多个chunk读取。在读完chunk的数据之前，chunk的数据会先缓存在内存中。每个chunk包含的行数越少，则表中的chunk的总数量越大，尽管这会降低故障恢复的粒度，但可能导致内存OOM和整体的吞吐量降低。因此，您需要进行权衡，并设置合理的chunk大小。
scan.snapshot.fetch.size	当读取表的全量数据时，每次最多拉取的记录数。	否	INTEGER	1024	无。
scan.startup.mode	消费数据时的启动模式。	否	STRING	initial	参数取值如下： initial （默认）：在第一次启动时，会先扫描历史全量数据，然后读取最新的Binlog数据。 latest-offset ：在第一次启动时，不会扫描历史全量数据，直接从Binlog的末尾（最新的Binlog处）开始读取，即只读取该连接器启动以后的最新变更。 earliest-offset ：不扫描历史全量数据，直接从可读取的最早Binlog开始读取。 specific-offset ：不扫描历史全量数据，从您指定的Binlog位点启动，位点可通过同时配置 scan.startup.specific-offset.file 和 scan.startup.specific-offset.pos 参数来指定从特定Binlog文件名和偏移量启动，也可以只配置 scan.startup.specific-offset.gtid-set 来指定从某个GTID集合启动。 timestamp ：不扫描历史全量数据，从指定的时间戳开始读取Binlog。时间戳通过 scan.startup.timestamp-millis 指定，单位为毫秒。
scan.startup.specific-offset.file	使用指定位点模式启动时，启动位点的Binlog文件名。	否	STRING	无	使用该配置时， scan.startup.mode 必须配置为 specific-offset 。文件名格式例如 `mysql-bin.000003` 。
scan.startup.specific-offset.pos	使用指定位点模式启动时，启动位点在指定Binlog文件中的偏移量。	否	INTEGER	无	使用该配置时， scan.startup.mode 必须配置为 specific-offset 。
scan.startup.specific-offset.gtid-set	使用指定位点模式启动时，启动位点的GTID集合。	否	STRING	无	使用该配置时， scan.startup.mode 必须配置为 specific-offset 。GTID集合格式例如 `24DA167-0C0C-11E8-8442-00059A3C7B00:1-19` 。
scan.startup.timestamp-millis	使用指定时间模式启动时，启动位点的毫秒时间戳。	否	LONG	无	使用该配置时， scan.startup.mode 必须配置为 timestamp 。时间戳单位为毫秒。
server-time-zone	数据库在使用的会话时区。	VVR-6.0.2以下版本必填，其他版本选填	STRING	如果您没有指定该参数，则系统默认使用Flink作业运行时的环境时区作为数据库服务器时区，即您选择的可用区所在的时区。	例如Asia/Shanghai，该参数控制了MySQL中的TIMESTAMP类型如何转成STRING类型。更多信息请参见 Debezium时间类型。
debezium.min.row.count.to.stream.results	当表的条数大于该值时，会使用分批读取模式。	否	INTEGER	1000	Flink采用以下方式读取MySQL源表数据：全量读取：直接将整个表的数据读取到内存里。优点是速度快，缺点是会消耗对应大小的内存，如果源表数据量非常大，可能会有OOM风险。分批读取：分多次读取，每次读取一定数量的行数，直到读取完所有数据。优点是读取数据量比较大的表没有OOM风险，缺点是读取速度相对较慢。
connect.timeout	连接MySQL数据库服务器超时时，重试连接之前等待超时的最长时间。	否	DURATION	30s	无。
connect.max-retries	连接MySQL数据库服务时，连接失败后重试的最大次数。	否	INTEGER	3	无。
connection.pool.size	数据库连接池大小。	否	INTEGER	20	数据库连接池用于复用连接，可以降低数据库连接数量。
jdbc.properties.*	JDBC URL中的自定义连接参数。	否	STRING	无	您可以传递自定义的连接参数，例如不使用SSL协议，则可配置为 'jdbc.properties.useSSL' = 'false' 。支持的连接参数请参见 MySQL Configuration Properties 。
debezium.*	Debezium读取Binlog的自定义参数。	否	STRING	无	您可以传递自定义的Debezium参数，例如使用'debezium.event.deserialization.failure.handling.mode'='ignore'来指定解析错误时的处理逻辑。
heartbeat.interval	Source通过心跳事件推动Binlog位点前进的时间间隔。	否	DURATION	30s	心跳事件用于推动Source中的Binlog位点前进，这对MySQL中更新缓慢的表非常有用。对于更新缓慢的表，Binlog位点无法自动前进，通过够心跳事件可以推到Binlog位点前进，可以避免Binlog位点不前进引起Binlog位点过期问题，Binlog位点过期会导致作业失败无法恢复，只能无状态重启。
scan.incremental.snapshot.chunk.key-column	可以指定某一列作为快照阶段切分分片的切分列。	见备注列。	STRING	无	无主键表必填，选择的列必须是非空类型（NOT NULL）。有主键的表为选填，仅支持从主键中选择一列。
rds.region-id	RDS实例所在的地域 ID。	使用读取OSS归档日志功能时必填。	STRING	无	仅Flink计算引擎VVR 6.0.7及以上版本支持。地域ID请参见地域和可用区。
rds.access-key-id	阿里云账号Access Key ID。	使用读取OSS归档日志功能时必填。	STRING	无	详情请参见如何查看AccessKey ID和AccessKey Secret信息？
rds.access-key-secret	阿里云账号Access Key Secret。	使用读取OSS归档日志功能时必填。	STRING	无	详情请参见如何查看AccessKey ID和AccessKey Secret信息？
rds.db-instance-id	RDS实例ID。	使用读取OSS归档日志功能时必填。	STRING	无	仅Flink计算引擎VVR 6.0.7及以上版本支持。
scan.incremental.close-idle-reader.enabled	是否在快照结束后关闭空闲的 Reader。	否	BOOLEAN	false	仅Flink计算引擎VVR 8.0.1及以上版本支持。该配置生效需要设置execution.checkpointing.checkpoints-after-tasks-finish.enabled为true。

参数	说明	是否必填	数据类型	默认值	备注
url	MySQL JDBC URL	否	STRING	无	URL的格式为： `jdbc:mysql://<连接地址>:<端口号>/<数据库名称>` 。
lookup.max-retries	读取数据失败后，重试读取的最大次数。	否	INTEGER	3	仅Flink计算引擎VVR 6.0.7及以上版本支持。
lookup.cache.strategy	缓存策略。	否	STRING	None	支持None、LRU和ALL三种缓存策略，取值含义详情请参见背景信息。
lookup.cache.max-rows	最大缓存条数。	否	INTEGER	100000	当选择LRU缓存策略后，必须设置缓存大小。当选择ALL缓存策略后，可以不设置缓存大小。
lookup.cache.ttl	缓存超时时间。	否	DURATION	10 s	lookup.cache.ttl 的配置和 lookup.cache.strategy 有关，详情如下：如果 lookup.cache.strategy 配置为 None ，则 lookup.cache.ttl 可以不配置，表示缓存不超时。如果 lookup.cache.strategy 配置为 LRU ，则 lookup.cache.ttl 为缓存超时时间。默认不过期。如果 lookup.cache.strategy 配置为 ALL ，则 lookup.cache.ttl 为缓存加载时间。默认不重新加载。填写时请使用时间格式，例如1min或10s。
lookup.max-join-rows	主表中每一条数据查询维表时，匹配后最多返回的结果数。	否	INTEGER	1024	在Flink计算引擎VVR 6.0.7及以上版本支持。

参数	说明	是否必填	数据类型	默认值	备注
url	MySQL JDBC URL	否	STRING	无	URL的格式为： `jdbc:mysql://<连接地址>:<端口号>/<数据库名称>` 。
sink.max-retries	写入数据失败后，重试写入的最大次数。	否	INTEGER	3	无。
sink.buffer-flush.batch-size	一次批量写入的条数。	否	INTEGER	4096	在Flink计算引擎VVR 6.0.7及以上版本支持。
sink.buffer-flush.max-rows	内存中缓存的数据条数。	否	INTEGER	在Flink计算引擎VVR 6.0.7版本以下，该参数默认值为100。在Flink计算引擎VVR 6.0.7版本及以上版本，该参数默认值为10000。	需指定主键后，该参数才生效。
sink.buffer-flush.interval	清空缓存的时间间隔。表示如果缓存中的数据在等待指定时间后，依然没有达到输出条件，系统会自动输出缓存中的所有数据。	否	DURATION	1s	无。
sink.ignore-delete	是否忽略数据Delete操作。	否	BOOLEAN	false	在Flink计算引擎VVR 6.0.7及以上版本支持。 Flink SQL可能会生成数据Delete操作，在多个输出节点根据主键同时更新同一张结果表的不同字段的场景下，可能导致数据结果不正确。例如一个任务在删除了一条数据后，另一个任务又只更新了这条数据的部分字段，其余未被更新的字段由于被删除，其值会变成null或默认值。通过将ignoreDelete设置为true，可以避免数据删除操作。
sink.ignore-null-when-update	更新数据时，如果传入的数据字段值为null，是更新对应字段为null，还是跳过该字段的更新。	否	BOOLEAN	false	参数取值如下： true：不更新该字段。指定主键后，该参数才可以设置为true，此时结果表写入数据不会攒批执行。 false：更新该字段为null。

MySQL CDC字段类型	Flink字段类型
TINYINT	TINYINT
SMALLINT	SMALLINT
TINYINT UNSIGNED
TINYINT UNSIGNED ZEROFILL
INT	INT
MEDIUMINT
SMALLINT UNSIGNED
SMALLINT UNSIGNED ZEROFILL
BIGINT	BIGINT
INT UNSIGNED
INT UNSIGNED ZEROFILL
MEDIUMINT UNSIGNED
MEDIUMINT UNSIGNED ZEROFILL
BIGINT UNSIGNED	DECIMAL(20, 0)
BIGINT UNSIGNED ZEROFILL
SERIAL
FLOAT [UNSIGNED] [ZEROFILL]	FLOAT
DOUBLE [UNSIGNED] [ZEROFILL]	DOUBLE
DOUBLE PRECISION [UNSIGNED] [ZEROFILL]
REAL [UNSIGNED] [ZEROFILL]
NUMERIC(p, s) [UNSIGNED] [ZEROFILL]	DECIMAL(p, s)
DECIMAL(p, s) [UNSIGNED] [ZEROFILL]	DECIMAL(p, s)
BOOLEAN	BOOLEAN
TINYINT(1)	BOOLEAN
DATE	DATE
TIME [(p)]	TIME [(p)] [WITHOUT TIME ZONE]
DATETIME [(p)]	TIMESTAMP [(p)] [WITHOUT TIME ZONE]
TIMESTAMP [(p)]	TIMESTAMP [(p)]
TIMESTAMP [(p)]	TIMESTAMP [(p)] WITH LOCAL TIME ZONE
CHAR(n)	STRING
VARCHAR(n)
TEXT
BINARY	BYTES
VARBINARY
BLOB

MySQL字段类型	Flink字段类型
TINYINT	TINYINT
SMALLINT	SMALLINT
TINYINT UNSIGNED	SMALLINT
INT	INT
MEDIUMINT
SMALLINT UNSIGNED
BIGINT	BIGINT
INT UNSIGNED	BIGINT
BIGINT UNSIGNED	DECIMAL(20, 0)
FLOAT	FLOAT
DOUBLE	DOUBLE
DOUBLE PRECISION	DOUBLE
NUMERIC(p, s)	DECIMAL(p, s)
DECIMAL(p, s)	DECIMAL(p, s)
BOOLEAN	BOOLEAN
TINYINT(1)	BOOLEAN
DATE	DATE
TIME [(p)]	TIME [(p)] [WITHOUT TIME ZONE]
DATETIME [(p)]	TIMESTAMP [(p)] [WITHOUT TIME ZONE]
TIMESTAMP [(p)]	TIMESTAMP [(p)] [WITHOUT TIME ZONE]
CHAR(n)	CHAR(n)
VARCHAR(n)	VARCHAR(n)
BIT(n)	BINARY(⌈n/8⌉)
BINARY(n)	BINARY(n)
VARBINARY(N)	VARBINARY(N)
TINYTEXT	STRING
TEXT
MEDIUMTEXT
LONGTEXT
TINYBLOB	BYTES
BLOB
MEDIUMBLOB
LONGBLOB

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;

元数据key	元数据类型	描述
database_name	STRING NOT NULL	包含该行记录的库名。
table_name	STRING NOT NULL	包含该行记录的表名。
op_ts	TIMESTAMP_LTZ(3) NOT NULL	该行记录在数据库中的变更时间，如果该记录来自表的存量历史数据而不是Binlog中获取，则该值总是0。

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;

```
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)' -- 正则表达式匹配多张表。
);
```
```
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' -- 在时间戳启动模式下指定启动时间戳。
)
```

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);
      .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");

参数	说明
hostname	MySQL数据库的IP地址或者Hostname。
port	MySQL数据库服务的端口号。
databaseList	MySQL数据库名称。
username	MySQL数据库服务的用户名。
password	MySQL数据库服务的密码。
deserializer	反序列化器，将SourceRecord类型记录反序列化到指定类型。参数取值如下： RowDataDebeziumDeserializeSchema：将SourceRecord转成Flink Table或SQL内部数据结构RowData。 JsonDebeziumDeserializationSchema：将SourceRecord转成JSON格式的String。

背景信息

特色功能

前提条件

使用限制

注意事项

语法结构

WITH参数

类型映射

使用示例

关于MySQL CDC源表

MySQL CDC DataStream API

常见问题