SQL

MySQL连接器可以在SQL作业中使用，作为源表，维表或者结果表。

语法结构

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>'
);

WITH参数

• 通用

  参数	说明	是否必填	数据类型	默认值	备注
  connector	表类型。	是	STRING	无	作为源表时，可以填写为 mysql-cdc 或者 mysql ，二者等价。作为维表或结果表时，固定值为 mysql 。
  hostname	MySQL数据库的IP地址或者Hostname。	是	STRING	无	建议填写专有网络VPC地址。 说明 如果MySQL与Flink全托管不在同一VPC，需要先打通跨VPC的网络或者使用公网的形式访问，详情请参见 如何访问跨VPC的其他服务？ 和 Flink全托管集群如何访问公网？ 。
  username	MySQL数据库服务的用户名。	是	STRING	无	无。
  password	MySQL数据库服务的密码。	是	STRING	无	无。
  database-name	MySQL数据库名称。	是	STRING	无	作为源表时，数据库名称支持正则表达式以读取多个数据库的数据。 使用正则表达式时，尽量不要使用 ^ 和 $ 符号匹配开头和结尾。具体原因详见 table-name 备注的说明。
  table-name	MySQL表名。	是	STRING	无	作为源表时，表名支持正则表达式以读取多个表的数据。 在读取多个MySQL表时，将多个CTAS语句作为一个作业提交，可以避免启用多个Binlog监听，提高性能和效率。详情请参见 示例四：多个CTAS语句作为一个作业提交 。 使用正则表达式时，尽量不要使用 ^ 和 $ 符号匹配开头和结尾。具体原因详见以下说明。 说明 MySQL CDC源表在正则匹配表名时，会将您填写的 database-name ， table-name 通过字符串 \\. （VVR 8.0.1前使用字符 . ）连接成为一个全路径的正则表达式，然后使用该正则表达式和MySQL数据库中表的全限定名进行正则匹配。 例如：当配置'database-name'='db_.*'且'table-name'='tb_.+'时，连接器将会使用正则表达式 db_.*\\.tb_.+ （8.0.1版本前为 db_.*.tb_.+ ）去匹配表的全限定名来确定需要读取的表。
  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 指定，单位为毫秒。 重要 earliest-offset ， specific-offset 和 timestamp 启动模式在Flink计算引擎VVR 6.0.4及以上的版本支持使用。 对于 earliest-offset ， specific-offset 和 timestamp 启动模式，如果启动时刻和指定的启动位点时刻的表结构不同，作业会因为表结构不同而报错。换一句话说，使用这三种启动模式，需要保证在指定的Binlog消费位置到作业启动的时间之间，对应表不能发生表结构变更。
  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 。时间戳单位为毫秒。 重要 在使用指定时间时，MySQL CDC会尝试读取每个Binlog文件的初始事件以确定其时间戳，最终定位至指定时间对应的Binlog文件。请保证指定的时间戳对应的Binlog文件在数据库上没有被清理且可以被读取到。
  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）。 有主键的表为选填，仅支持从主键中选择一列。 说明 仅Flink计算引擎VVR 6.0.7及以上版本支持。
  rds.region-id	阿里云RDS MySQL实例所在的地域ID。	使用读取OSS归档日志功能时必填。	STRING	无	仅Flink计算引擎VVR 6.0.7及以上版本支持。 地域ID请参见 地域和可用区 。
  rds.access-key-id	阿里云RDS MySQL账号Access Key ID。	使用读取OSS归档日志功能时必填。	STRING	无	详情请参见 如何查看AccessKey ID和AccessKey Secret信息？ 重要 仅Flink计算引擎VVR 6.0.7及以上版本支持。 为了避免您的AK信息泄露，建议您通过密钥管理的方式填写AccessKey ID取值，详情请参见 变量管理 。
  rds.access-key-secret	阿里云RDS MySQL账号Access Key Secret。	使用读取OSS归档日志功能时必填。	STRING	无	详情请参见 如何查看AccessKey ID和AccessKey Secret信息？ 重要 仅Flink计算引擎VVR 6.0.7及以上版本支持。 为了避免您的AK信息泄露，建议您通过密钥管理的方式填写AccessKey Secret取值，详情请参见 变量管理 。
  rds.db-instance-id	阿里云RDS MySQL实例ID。	使用读取OSS归档日志功能时必填。	STRING	无	仅Flink计算引擎VVR 6.0.7及以上版本支持。
  rds.main-db-id	阿里云RDS MySQL实例主库编号。	否	STRING	无	获取主库编号详情请参见 RDS MySQL日志备份 。 仅Flink计算引擎VVR 8.0.7及以上版本支持。
  rds.download.timeout	从OSS下载单个归档日志的超时时间。	否	DURATION	60s	仅Flink计算引擎VVR 6.0.7及以上版本支持。
  rds.endpoint	获取OSS Binlog信息的服务接入点。	否	STRING	无	可选值详情请参见 服务接入点 。 仅Flink计算引擎VVR 8.0.8及以上版本支持。
  scan.incremental.close-idle-reader.enabled	是否在快照结束后关闭空闲的 Reader。	否	BOOLEAN	false	仅Flink计算引擎VVR 8.0.1及以上版本支持。 该配置生效需要设置execution.checkpointing.checkpoints-after-tasks-finish.enabled为true。
  scan.read-changelog-as-append-only.enabled	是否将changelog数据流转换为append-only数据流。	否	BOOLEAN	false	参数取值如下： true ：所有类型的消息（包括INSERT、DELETE、UPDATE_BEFORE、UPDATE_AFTER）都会转换成INSERT类型的消息。仅在需要保存上游表删除消息等特殊场景下开启使用。 false （默认）：所有类型的消息都保持原样下发。 说明 仅Flink计算引擎VVR 8.0.8及以上版本支持。
  scan.only.deserialize.captured.tables.changelog.enabled	在增量阶段，是否仅对指定表的变更事件进行反序列化。	否	BOOLEAN	false	参数取值如下： true ：仅对目标表的变更数据进行反序列化，加快Binlog读取速度。 false （默认）：对所有表的变更数据进行反序列化。 说明 仅Flink计算引擎VVR 8.0.7及以上版本支持。 在Flink计算引擎VVR 8.0.8及以下版本使用时，参数名需要修改为 debezium.scan.only.deserialize.captured.tables.changelog.enable 。
  scan.parallel-deserialize-changelog.enabled	在增量阶段,是否使用多线程对变更事件进行解析。	否	BOOLEAN	false	参数取值如下： true ：在变更事件的反序列化阶段采用多线程处理，同时保证Binlog事件顺序不变，从而加快读取速度。 false （默认）：在事件的反序列化阶段使用单线程处理。 说明 仅Flink计算引擎VVR 8.0.7及以上版本支持。

• 维表独有

  参数	说明	是否必填	数据类型	默认值	备注
  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三种缓存策略，取值含义详情请参见 背景信息 。 说明 使用LRU缓存策略时，还必须配置 lookup.cache.max-rows 参数。
  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及以上版本支持。
  lookup.filter-push-down.enabled	是否开启维表Filter下推。	否	BOOLEAN	false	参数取值如下： true ：开启维表Filter下推，在加载MySQL数据库表的数据时，维表会根据SQL作业中设置的条件提前过滤数据。 false （默认）：不开启维表Filter下推，在加载MySQL数据库表的数据时，维表会加载全量数据。 说明 仅Flink计算引擎VVR 8.0.7及以上版本支持。 重要 维表下推应该仅在Flink表用作维表时开启。MySQL源表暂不支持开启Filter下推，如果一张Flink表同时被作为源表和维表，且维表开启了Filter下推，则在使用源表处需要通过 SQL Hints 的方式将该配置项显式设为false，否则可能导致作业运行异常。

• 结果表独有

  参数	说明	是否必填	数据类型	默认值	备注
  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或默认值。通过将sink.ignore-delete设置为true，可以避免数据删除操作。
  sink.ignore-null-when-update	更新数据时，如果传入的数据字段值为null，是更新对应字段为null，还是跳过该字段的更新。	否	BOOLEAN	false	参数取值如下： true：不更新该字段。但是当Flink表设置主键时，才支持配置该参数为true。配置为true时： 如果是8.0.6及以下的版本，结果表写入数据不支持攒批执行。 如果是8.0.7及以上的版本，结果表写入数据支持攒批执行。 攒批写入虽然可以明显增强写入效率和整体吞吐量，但是会带来数据延迟问题和内存溢出风险。因此请您根据实际业务场景做好权衡。 false：更新该字段为null。 说明 仅实时计算引擎VVR 8.0.5及以上版本支持该参数。

类型映射

• CDC源表

  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]
  BOOLEAN	BOOLEAN
  TINYINT(1)
  DATE	DATE
  TIME [(p)]	TIME [(p)] [WITHOUT TIME ZONE]
  DATETIME [(p)]	TIMESTAMP [(p)] [WITHOUT TIME ZONE]
  TIMESTAMP [(p)]	TIMESTAMP [(p)]
  	TIMESTAMP [(p)] WITH LOCAL TIME ZONE
  CHAR(n)	STRING
  VARCHAR(n)
  TEXT
  BINARY	BYTES
  VARBINARY
  BLOB

  重要

  建议MySQL不要使用TINYINT(1)类型存储0和1以外的数值，当 property-version =0时，默认MySQL CDC源表会将TINYINT(1)映射到Flink的BOOLEAN上，造成数据不准确。如果需要使用TINYINT(1)类型存储0和1以外的数值，请参见 MySQL Catalog参数配置 。

• 维表和结果表

  MySQL字段类型	Flink字段类型
  TINYINT	TINYINT
  SMALLINT	SMALLINT
  TINYINT UNSIGNED
  INT	INT
  MEDIUMINT
  SMALLINT UNSIGNED
  BIGINT	BIGINT
  INT UNSIGNED
  BIGINT UNSIGNED	DECIMAL(20, 0)
  FLOAT	FLOAT
  DOUBLE	DOUBLE
  DOUBLE PRECISION
  NUMERIC(p, s)	DECIMAL(p, s) 说明 其中p <= 38。
  DECIMAL(p, s)
  BOOLEAN	BOOLEAN
  TINYINT(1)
  DATE	DATE
  TIME [(p)]	TIME [(p)] [WITHOUT TIME ZONE]
  DATETIME [(p)]	TIMESTAMP [(p)] [WITHOUT TIME ZONE]
  TIMESTAMP [(p)]
  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 重要 Flink仅支持小于等于2,147,483,647(2^31 - 1)的MySQL BLOB类型的记录。
  BLOB
  MEDIUMBLOB
  LONGBLOB

数据摄入

MySQL连接器作为数据源可以在数据摄入YAML作业中使用。

语法结构

source:
   type: mysql
   name: MySQL Source
   hostname: localhost
   port: 3306
   username: <username>
   password: <password>
   tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
   server-id: 5401-5404
sink:
  type: xxx

配置项

类型映射

数据摄入类型映射如下表所示。