流计算 Oceanus MySQL 连接器

MySQL 连接器支持从 MySQL 数据库读取快照数据与增量 binlog 数据，并提供整库同步、表结构变更同步等能力。
它同时可以作为 Source（type: mysql）与 Sink（type: mysql）使用，通过 YAML 作业定义即可完成一条数据同步链路。
﻿
前置准备
使用 MySQL Source 前，需要确保 MySQL 实例满足下列条件：
﻿
binlog 已开启，且 binlog_format = ROW；建议同时设置 binlog_row_image = FULL，以保证更新前后镜像完整。
数据库账号具备读取 binlog 与快照数据所需的权限，至少包含 SELECT、REPLICATION SLAVE、REPLICATION CLIENT、SHOW DATABASES、RELOAD。
MySQL 版本建议使用 5.7 或 8.0；若使用 GTID 作为启动位点，需确保实例已开启 GTID。
每个接入 MySQL 集群的作业都必须使用独占的 server-id，多个作业之间以及与现有从库之间都不能重复。
﻿
版本说明
Flink 版本
说明
1.20
基于 Flink-cdc 社区 release-3.6 开发，JDK 版本为 11，支持 MySQL 版本为 5.6, 5.7, 8.0.x，8.4+
使用范围
 可作为 Source（type: mysql）与 Sink（type: mysql）使用
﻿
快速示例
MySQL 同步到 Doris
source:
  type: mysql
  name: MySQL Source
  hostname: 127.0.0.1
  port: 3306
  username: admin
  password: pass
  tables: adb.\\.*, bdb.user_table_[0-9]+, [app|web].order_\\.*
  server-id: 5401-5404
  server-time-zone: Asia/Shanghai
﻿
sink:
  type: doris
  name: Doris Sink
  fenodes: 127.0.0.1:8030
  username: root
  password: pass
﻿
pipeline:
  name: MySQL to Doris Pipeline
  parallelism: 4
﻿
MySQL 同步到 MySQL
source:
  type: mysql
  name: MySQL Source
  hostname: 127.0.0.1
  port: 3306
  username: admin
  password: pass
  tables: adb.\\.*
  server-id: 5401-5404
﻿
sink:
  type: mysql
  name: MySQL Sink
  jdbc-url: jdbc:mysql://127.0.0.1:3306/adb_sink
  username: root
  password: pass
  sink.buffer-flush.interval: 2s
  sink.buffer-flush.max-rows: 5000
﻿
pipeline:
  name: MySQL to MySQL Pipeline
  parallelism: 4
﻿
Source 配置项
常用配置项
参数
是否必填
默认值
类型
说明
hostname
required
(none)
String
MySQL 数据库服务器的 IP 地址或主机名。
port
optional
3306
Integer
MySQL 数据库服务器的端口号。
username
required
(none)
String
连接 MySQL 数据库服务器时使用的用户名。
password
required
(none)
String
连接 MySQL 数据库服务器时使用的密码。
tables
required
(none)
String
需要捕获的表名，格式为 库名.表名，支持正则表达式，多条规则用英文逗号分隔。
注意：点号（.）是库名与表名的分隔符，如果需要在正则表达式中用点号匹配任意字符，必须用反斜杠转义。
例如：db0.\\.*, db1.user_table_[0-9]+, db[1-2].[app|web]_order_\\.*
若按该规则匹配不到任何表，作业启动时会直接报错。
tables.exclude
optional
(none)
String
需要排除的表名，在 tables 匹配结果之上做二次排除，用法与 tables 相同。
server-time-zone
optional
(none)
String
数据库服务器的会话时区，例如 Asia/Shanghai。未配置时使用作业运行环境的系统时区， 对时间相关字段可能造成数据不一致，建议显式配置。
server-id
optional
(none)
String
本作业在 MySQL 集群中的 slave id，语法支持单个整数（5400）或整数范围（5400-5408）。
建议配置为整数范围，且必须保证与集群中其它进程（含已有从库和其它 CDC 作业）不冲突。
未配置时会在 6401 ~ 2147483647 之间随机生成，生产环境建议显式指定。
schema-change.enabled
optional
true
Boolean
是否发送表结构变更事件。开启后下游 Sink 才能感知并同步表结构变更。
scan.incremental.snapshot.chunk.size
optional
8096
Integer
快照阶段单个 chunk 的行数。快照读取时表会被切分为多个 chunk。
scan.snapshot.fetch.size
optional
1024
Integer
读取表快照时每次拉取的最大行数。开启快照快速通道后，显式配置该值会使用服务端游标并按此值分批拉取； 未显式配置时快速通道使用 Connector/J 的流式读取。
scan.startup.mode
optional
initial
String
启动模式，可取 initial、earliest-offset、latest-offset、 specific-offset、timestamp、snapshot，详见「启动模式」章节。
scan.startup.specific-offset.file
optional
(none)
String
specific-offset 启动模式下的 binlog 文件名。
scan.startup.specific-offset.pos
optional
(none)
Long
specific-offset 启动模式下的 binlog 位点偏移量。
scan.startup.specific-offset.gtid-set
optional
(none)
String
specific-offset 启动模式下的 GTID 集合。
scan.startup.specific-offset.skip-events
optional
(none)
Long
从指定启动位点开始后需要跳过的事件数量。
scan.startup.specific-offset.skip-rows
optional
(none)
Long
从指定启动位点开始后需要跳过的数据行数。
scan.startup.timestamp-millis
optional
(none)
Long
timestamp 启动模式下的毫秒时间戳。
connect.timeout
optional
30s
Duration
连接 MySQL 服务器的超时时间。
connect.max-retries
optional
3
Integer
建立 MySQL 连接的最大重试次数。
connection.pool.size
optional
20
Integer
连接池大小。
heartbeat.interval
optional
30s
Duration
发送心跳事件的间隔，用于跟踪最新可用的 binlog 位点。
jdbc.properties.*
optional
(none)
String
透传给 JDBC URL 的自定义属性，例如 jdbc.properties.useSSL: 'false'。
debezium.*
optional
(none)
String
透传给 Debezium Embedded Engine 的属性，例如 debezium.snapshot.mode: never。
﻿
高级配置
下列参数用于性能调优和特殊场景，请结合实际场景使用。
参数
默认值
类型
说明
chunk-meta.group.size
1000
Integer
chunk 元数据的分组大小，超过该大小时元数据会被拆成多个分组。
chunk-key.even-distribution.factor.upper-bound
1000.0
Double
chunk 切分键分布因子的上界。分布因子按 (MAX(id) - MIN(id) + 1) / rowCount 计算， 用于判断表数据是否均匀分布：分布均匀时按算法直接切分 chunk，不均匀时才回查 MySQL 进行切分。 取值必须大于等于 1.0。
chunk-key.even-distribution.factor.lower-bound
0.05
Double
chunk 切分键分布因子的下界，取值必须落在 [0, 1] 区间内。
scan.incremental.snapshot.chunk.key-column
(none)
String
指定快照切分使用的列，默认取主键的第一列。
格式为 库名.表名:列名，多张表之间用分号分隔， 例如 db1.user_table_[0-9]+:col1;db[1-2].[app|web]_order_\\.*:col2。
scan.incremental.close-idle-reader.enabled
false
Boolean
快照结束后是否关闭空闲的 Reader。依赖 FLIP-147（Checkpoints After Tasks Finished）， Flink 版本需大于等于 1.14。
scan.newly-added-table.enabled
false
Boolean
是否开启动态加表。从 savepoint/checkpoint 恢复时，对新增表重新做全量快照并继续读取增量。
不能与 scan.binlog.newly-added-table.enabled 同时开启。
scan.binlog.newly-added-table.enabled
false
Boolean
增量阶段是否读取新增表的 DDL 与 DML，只做增量读取、不做重新快照。
不能与 scan.newly-added-table.enabled 同时开启。
metadata.list
(none)
String
需要额外透传到下游的元数据列表，英文逗号分隔，当前可用值为 op_ts。 op_ts 为大整型（BIGINT NOT NULL）的变更时间戳，可直接在 transform 中引用。
scan.parse.online.schema.changes.enabled
false
Boolean
是否解析 gh-ost / pt-osc 等在线变更工具产生的表结构变更事件。 这类工具会先在「影子表」上执行变更再与原表交换。旧参数名 gh-ost.ddl.change 仍可兼容，建议使用本参数。
include-comments.enabled
false
Boolean
是否同步表和字段的注释。开启后会带来额外的内存开销。
treat-tinyint1-as-boolean.enabled
true
Boolean
是否将 TINYINT(1) 当作 BOOLEAN 处理。关闭后会设置 JDBC 属性 tinyInt1isBit=false， 此时 TINYINT(1) 映射为 TINYINT。
use.legacy.json.format
true
Boolean
是否使用 legacy JSON 格式解析 binlog 中的 JSON 数据。 为 true 时去掉值前与逗号后的空格，例如 {"key1": "value1"} 会输出为 {"key1":"value1"}； 为 false 时保留原始空格。
scan.incremental.snapshot.unbounded-chunk-first.enabled
false
Boolean
快照阶段是否优先分配 unbounded chunk，有助于降低 TaskManager 在快照阶段发生 OOM 的风险。
scan.incremental.snapshot.backfill.skip
false
Boolean
快照阶段是否跳过 backfill。跳过之后，快照期间发生的变更会在增量阶段回放，而不是合并进快照结果。
注意：跳过 backfill 可能带来数据不一致，只能保证 at-least-once 语义。
scan.snapshot.fast-path.enabled
false
Boolean
是否开启快照快速通道：直接由 JDBC 结果集转换记录，不再构造 Debezium SourceRecord。
该参数必须与 scan.incremental.snapshot.backfill.skip=true 同时配置，否则作业启动校验失败。
scan.snapshot.fast-path.batch-size
1024
Integer
快照快速通道下每批转换的记录条数。
scan.weight-string-compare.enabled
false
Boolean
切分键为 varchar 类型时，是否使用 WEIGHT_STRING 进行比较。
scan.only.deserialize.captured.tables.changelog.enabled
false
Boolean
增量阶段是否只反序列化被捕获表的变更事件。开启后可加快 binlog 读取速度。
scan.case-insensitive-column-name.enabled
true
Boolean
数据库大小写不敏感时，是否将列名统一转换为小写。关闭后保留列名的原始大小写。
scan.idle.latency.reset.enabled
false
Boolean
Source 空闲超过指定时长后，是否把 currentEmitEventTimeLag 与 currentFetchEventTimeLag 指标归零。消费历史 binlog 追平后，该指标否则会一直停留在较大数值。
scan.idle.latency.reset.timeout.ms
300000
Long
判定 Source 空闲的时长阈值（毫秒），仅当 scan.idle.latency.reset.enabled=true 时生效。
﻿
启动模式
scan.startup.mode 决定作业从哪里开始读取数据，取值如下（取值大小写不敏感）：
﻿
initial（默认）：首次启动时先对捕获的表做全量快照，快照完成后继续读取最新 binlog。
earliest-offset：跳过快照，从可读取的最早 binlog 位点开始读取。
latest-offset：跳过快照，只从 binlog 末尾开始读取，即只消费作业启动之后产生的变更。
specific-offset：跳过快照，从指定位点开始读取。位点可以用 binlog 文件名加位置指定，也可以用 GTID 集合指定。
timestamp：跳过快照，从指定的时间戳开始读取 binlog 事件。
snapshot：只做快照阶段，不进入增量阶段，快照读取完成后作业退出。
﻿
从指定的 binlog 文件名与位点开始：
source:
  type: mysql
  scan.startup.mode: specific-offset
  scan.startup.specific-offset.file: 'mysql-bin.000003'
  scan.startup.specific-offset.pos: 4
﻿
从指定的 GTID 集合开始：
source:
  type: mysql
  scan.startup.mode: specific-offset
  scan.startup.specific-offset.gtid-set: '24DA167-0d...:1-19'
﻿
从指定的时间戳开始：
source:
  type: mysql
  scan.startup.mode: timestamp
  scan.startup.timestamp-millis: 1667232000000
﻿
specific-offset 模式下，启动位点有两种指定方式：scan.startup.specific-offset.gtid-set，或
（scan.startup.specific-offset.file + scan.startup.specific-offset.pos）。至少要配置其中一种，否则作业启动校验失败。
﻿
Sink 配置项
MySQL 作为 Sink 时的配置项如下：
参数
是否默认
默认值
类型
说明
jdbc-url
required
(none)
String
目标库的 JDBC 连接串，例如 jdbc:mysql://127.0.0.1:3306/db。
username
required
(none)
String
目标库的用户名。
password
required
(none)
String
目标库的密码。
driver-class-name
optional
org.apache.flink.cdc.connectors.shaded.com.mysql.cj.jdbc.Driver
String
JDBC 驱动类名，默认使用内置驱动，通常无需修改。
sink.buffer-flush.interval
optional
1s
Duration
缓冲区刷新间隔，超过该时长即冲刷缓存的记录。设为 0 表示关闭基于时间的刷新。
sink.buffer-flush.max-rows
optional
1000
Integer
触发刷新的最大缓存记录条数。
sink.max-retries
optional
3
Integer
写入 MySQL 失败时的最大重试次数。
append-only
optional
false
Boolean
是否以追加模式写入。追加模式下删除事件会被忽略、更新事件按普通插入写出，目标表可以不包含主键； 关闭时每张目标表都必须有主键。
Sink 侧的时区取自 pipeline 配置 local-time-zone，未配置时使用作业运行环境的系统时区。
﻿
Source 数据类型映射
MySQL 作为 Source 时，按下表把读取到的 MySQL 字段类型转换为 CDC 类型。
MySQL type
CDC type
NOTE
BIT(1)
BOOLEAN
﻿
BIT(n)，n > 1
BINARY(⌈(n + 7) / 8⌉)
﻿
BOOL
BOOLEAN
BOOLEAN
﻿
TINYINT(1)
BOOLEAN 或 TINYINT
受 treat-tinyint1-as-boolean.enabled 控制，默认映射为 BOOLEAN，关闭后映射为 TINYINT。
TINYINT(n)，n > 1
TINYINT
﻿
SMALLINT
TINYINT UNSIGNED
TINYINT UNSIGNED ZEROFILL
SMALLINT
﻿
INT
INTEGER
YEAR
MEDIUMINT
MEDIUMINT UNSIGNED
MEDIUMINT UNSIGNED ZEROFILL
SMALLINT UNSIGNED
SMALLINT UNSIGNED ZEROFILL
INT
﻿
BIGINT
INT UNSIGNED
INT UNSIGNED ZEROFILL
INTEGER UNSIGNED
INTEGER UNSIGNED ZEROFILL
BIGINT
﻿
BIGINT UNSIGNED
BIGINT UNSIGNED ZEROFILL
SERIAL
DECIMAL(20, 0)
﻿
FLOAT
FLOAT UNSIGNED
FLOAT UNSIGNED ZEROFILL
FLOAT
未显式指定精度时映射为 FLOAT。
FLOAT(p, s)
DOUBLE
显式指定了精度时按 DOUBLE 处理。
REAL
REAL UNSIGNED
REAL UNSIGNED ZEROFILL
DOUBLE
DOUBLE UNSIGNED
DOUBLE UNSIGNED ZEROFILL
DOUBLE PRECISION
DOUBLE PRECISION UNSIGNED
DOUBLE PRECISION UNSIGNED ZEROFILL
REAL(p, s)
DOUBLE(p, s)
DOUBLE
﻿
NUMERIC(p, s)
NUMERIC(p, s) UNSIGNED
NUMERIC(p, s) UNSIGNED ZEROFILL
DECIMAL(p, s)
DECIMAL(p, s) UNSIGNED
DECIMAL(p, s) UNSIGNED ZEROFILL
FIXED(p, s)
FIXED(p, s) UNSIGNED
FIXED(p, s) UNSIGNED ZEROFILL
其中 p <= 38
DECIMAL(p, s)
﻿
NUMERIC(p, s) 等 decimal 类型
其中 38 < p <= 65
STRING
MySQL 的 decimal 精度最大为 65，而 Flink decimal 精度上限为 38，因此超过 38 时映射为 STRING 以避免精度丢失。
DATE
DATE
﻿
TIME [(p)]
TIME [(p)]
﻿
DATETIME [(p)]
TIMESTAMP [(p)]
未指定精度时按 TIMESTAMP(0) 处理。
TIMESTAMP [(p)]
TIMESTAMP_LTZ [(p)]
未指定精度时按 TIMESTAMP_LTZ(0) 处理。
CHAR(n)
CHAR(n)
﻿
VARCHAR(n)
VARCHAR(n)
VARCHAR(0) 映射为 STRING。
BINARY(n)
BINARY(n)
﻿
VARBINARY(n)
VARBINARY(n)
﻿
TINYTEXT
TEXT
MEDIUMTEXT
LONGTEXT
LONG
LONG VARCHAR
JSON
ENUM
STRING
JSON 类型以 JSON 格式字符串的形式读出，格式受 use.legacy.json.format 影响。
TINYBLOB
BLOB
MEDIUMBLOB
LONGBLOB
BYTES
﻿
SET
ARRAY(STRING)
﻿
GEOMETRY
POINT
LINESTRING
POLYGON
MULTIPOINT
MULTILINESTRING
MULTIPOLYGON
GEOMETRYCOLLECTION
STRING
空间类型会转换为固定格式的 GeoJSON 字符串，详见「空间数据类型映射」章节。
未在上表中出现的字段类型暂不支持，同步包含此类字段的表会导致作业失败。
﻿
空间数据类型映射
除 GEOMETRYCOLLECTION 之外的空间类型都会被转换为如下固定格式的 JSON 字符串：
{"srid": 0 , "type": "xxx", "coordinates": [0, 0]}
﻿
srid 标识几何体所在的 SRS，未显式指定 SRID 时取 0。
type 标识空间数据类型，如 Point、LineString、Polygon。
coordinates 表示空间数据的坐标。
﻿
GEOMETRYCOLLECTION 会转换为如下格式：
{"srid": 0 , "type": "GeometryCollection", "geometries": [{"type":"Point","coordinates":[10,10]}]}
﻿
geometries 是包含所有空间数据的数组。
﻿
各类空间数据的转换示例如下：
Spatial data in MySQL
Json String converted in Flink
POINT(1 1)
{"coordinates":[1,1],"type":"Point","srid":0}
LINESTRING(3 0, 3 3, 3 5)
{"coordinates":[[3,0],[3,3],[3,5]],"type":"LineString","srid":0}
POLYGON((1 1, 2 1, 2 2, 1 2, 1 1))
{"coordinates":[[[1,1],[2,1],[2,2],[1,2],[1,1]]],"type":"Polygon","srid":0}
MULTIPOINT((1 1),(2 2))
{"coordinates":[[1,1],[2,2]],"type":"MultiPoint","srid":0}
MULTILINESTRING((1 1,2 2,3 3),(4 4,5 5))
{"coordinates":[[[1,1],[2,2],[3,3]],[[4,4],[5,5]]],"type":"MultiLineString","srid":0}
MULTIPOLYGON(((0 0, 10 0, 10 10, 0 10, 0 0)), ((5 5, 7 5, 7 7, 5 7, 5 5)))
{"coordinates":[[[[0,0],[10,0],[10,10],[0,10],[0,0]]],[[[5,5],[7,5],[7,7],[5,7],[5,5]]]],"type":"MultiPolygon","srid":0}
GEOMETRYCOLLECTION(POINT(10 10), POINT(30 30), LINESTRING(15 15, 20 20))
{"geometries":[{"type":"Point","coordinates":[10,10]},{"type":"Point","coordinates":[30,30]},{"type":"LineString","coordinates":[[15,15],[20,20]]}],"type":"GeometryCollection","srid":0}
﻿
Sink 数据类型映射
MySQL 作为 Sink 时，会自动为目标表生成 CREATE TABLE 语句（若表不存在），并按下列规则把 CDC 类型转换为目标库的 MySQL 字段类型。
CDC type
MySQL 目标字段类型
NOTE
BOOLEAN
TINYINT(1)
﻿
TINYINT
TINYINT
﻿
SMALLINT
SMALLINT
﻿
INT
INT
﻿
BIGINT
BIGINT
﻿
FLOAT
FLOAT
﻿
DOUBLE
DOUBLE
﻿
DECIMAL(p, s)
DECIMAL(p, s)
﻿
CHAR(n)
CHAR(n)，超长时降级为 VARCHAR / TEXT
n 不超过 255 时使用 CHAR(n)，超过时按 VARCHAR 的长度规则处理。
VARCHAR(n) / STRING
VARCHAR(n) / TEXT / MEDIUMTEXT / LONGTEXT
按字符长度自动选择：不超过 16383 用 VARCHAR(n)，不超过 65535 用 TEXT， 不超过 4194303 用 MEDIUMTEXT，更长用 LONGTEXT，保证目标字段一定装得下。
BINARY(n)
BINARY(n) / VARBINARY(n) / LONGBLOB
n 不超过 255 用 BINARY(n)，不超过 65535 用 VARBINARY(n)，更长用 LONGBLOB。
VARBINARY(n) / BYTES
VARBINARY(n) / LONGBLOB
n 不超过 65535 用 VARBINARY(n)，更长用 LONGBLOB。
DATE
DATE
﻿
TIME(p)
TIME(p)
小数秒精度超过 6 时按 6 处理。
TIMESTAMP(p)
DATETIME(p)
不带时区的时间戳写入 DATETIME，小数秒精度超过 6 时按 6 处理。
TIMESTAMP_LTZ(p)
DATETIME(p)
先按 pipeline 的 local-time-zone 换算成挂钟时间再写入 DATETIME， 因此落库结果不受 JVM 默认时区和服务端会话时区影响。
TIMESTAMP WITH TIME ZONE(p)
TIMESTAMP(p)
小数秒精度超过 6 时按 6 处理。
ARRAY
MAP
ROW
JSON
复合类型会序列化为 JSON 字符串后写入。ARRAY 内的二进制元素按 Base64 编码。
﻿

Flink 版本	说明
1.20	基于 Flink-cdc 社区 release-3.6 开发，JDK 版本为 11，支持 MySQL 版本为 5.6, 5.7, 8.0.x，8.4+

参数	是否必填	默认值	类型	说明
hostname	required	(none)	String	MySQL 数据库服务器的 IP 地址或主机名。
port	optional	3306	Integer	MySQL 数据库服务器的端口号。
username	required	(none)	String	连接 MySQL 数据库服务器时使用的用户名。
password	required	(none)	String	连接 MySQL 数据库服务器时使用的密码。
tables	required	(none)	String	需要捕获的表名，格式为库名.表名，支持正则表达式，多条规则用英文逗号分隔。注意：点号（.）是库名与表名的分隔符，如果需要在正则表达式中用点号匹配任意字符，必须用反斜杠转义。例如：db0.\\., db1.user_table_[0-9]+, db[1-2].[app\|web]_order_\\. 若按该规则匹配不到任何表，作业启动时会直接报错。
tables.exclude	optional	(none)	String	需要排除的表名，在 tables 匹配结果之上做二次排除，用法与 tables 相同。
server-time-zone	optional	(none)	String	数据库服务器的会话时区，例如 Asia/Shanghai。未配置时使用作业运行环境的系统时区，对时间相关字段可能造成数据不一致，建议显式配置。
server-id	optional	(none)	String	本作业在 MySQL 集群中的 slave id，语法支持单个整数（5400）或整数范围（5400-5408）。建议配置为整数范围，且必须保证与集群中其它进程（含已有从库和其它 CDC 作业）不冲突。未配置时会在 6401 ~ 2147483647 之间随机生成，生产环境建议显式指定。
schema-change.enabled	optional	true	Boolean	是否发送表结构变更事件。开启后下游 Sink 才能感知并同步表结构变更。
scan.incremental.snapshot.chunk.size	optional	8096	Integer	快照阶段单个 chunk 的行数。快照读取时表会被切分为多个 chunk。
scan.snapshot.fetch.size	optional	1024	Integer	读取表快照时每次拉取的最大行数。开启快照快速通道后，显式配置该值会使用服务端游标并按此值分批拉取；未显式配置时快速通道使用 Connector/J 的流式读取。
scan.startup.mode	optional	initial	String	启动模式，可取 initial、earliest-offset、latest-offset、 specific-offset、timestamp、snapshot，详见「启动模式」章节。
scan.startup.specific-offset.file	optional	(none)	String	specific-offset 启动模式下的 binlog 文件名。
scan.startup.specific-offset.pos	optional	(none)	Long	specific-offset 启动模式下的 binlog 位点偏移量。
scan.startup.specific-offset.gtid-set	optional	(none)	String	specific-offset 启动模式下的 GTID 集合。
scan.startup.specific-offset.skip-events	optional	(none)	Long	从指定启动位点开始后需要跳过的事件数量。
scan.startup.specific-offset.skip-rows	optional	(none)	Long	从指定启动位点开始后需要跳过的数据行数。
scan.startup.timestamp-millis	optional	(none)	Long	timestamp 启动模式下的毫秒时间戳。
connect.timeout	optional	30s	Duration	连接 MySQL 服务器的超时时间。
connect.max-retries	optional	3	Integer	建立 MySQL 连接的最大重试次数。
connection.pool.size	optional	20	Integer	连接池大小。
heartbeat.interval	optional	30s	Duration	发送心跳事件的间隔，用于跟踪最新可用的 binlog 位点。
jdbc.properties.*	optional	(none)	String	透传给 JDBC URL 的自定义属性，例如 jdbc.properties.useSSL: 'false'。
debezium.*	optional	(none)	String	透传给 Debezium Embedded Engine 的属性，例如 debezium.snapshot.mode: never。

参数	是否默认	默认值	类型	说明
jdbc-url	required	(none)	String	目标库的 JDBC 连接串，例如 jdbc:mysql://127.0.0.1:3306/db。
username	required	(none)	String	目标库的用户名。
password	required	(none)	String	目标库的密码。
driver-class-name	optional	org.apache.flink.cdc.connectors.shaded.com.mysql.cj.jdbc.Driver	String	JDBC 驱动类名，默认使用内置驱动，通常无需修改。
sink.buffer-flush.interval	optional	1s	Duration	缓冲区刷新间隔，超过该时长即冲刷缓存的记录。设为 0 表示关闭基于时间的刷新。
sink.buffer-flush.max-rows	optional	1000	Integer	触发刷新的最大缓存记录条数。
sink.max-retries	optional	3	Integer	写入 MySQL 失败时的最大重试次数。
append-only	optional	false	Boolean	是否以追加模式写入。追加模式下删除事件会被忽略、更新事件按普通插入写出，目标表可以不包含主键；关闭时每张目标表都必须有主键。

MySQL type	CDC type	NOTE
BIT(1)	BOOLEAN
BIT(n)，n > 1	BINARY(⌈(n + 7) / 8⌉)
BOOL BOOLEAN	BOOLEAN
TINYINT(1)	BOOLEAN 或 TINYINT	受 treat-tinyint1-as-boolean.enabled 控制，默认映射为 BOOLEAN，关闭后映射为 TINYINT。
TINYINT(n)，n > 1	TINYINT
SMALLINT TINYINT UNSIGNED TINYINT UNSIGNED ZEROFILL	SMALLINT
INT INTEGER YEAR MEDIUMINT MEDIUMINT UNSIGNED MEDIUMINT UNSIGNED ZEROFILL SMALLINT UNSIGNED SMALLINT UNSIGNED ZEROFILL	INT
BIGINT INT UNSIGNED INT UNSIGNED ZEROFILL INTEGER UNSIGNED INTEGER UNSIGNED ZEROFILL	BIGINT
BIGINT UNSIGNED BIGINT UNSIGNED ZEROFILL SERIAL	DECIMAL(20, 0)
FLOAT FLOAT UNSIGNED FLOAT UNSIGNED ZEROFILL	FLOAT	未显式指定精度时映射为 FLOAT。
FLOAT(p, s)	DOUBLE	显式指定了精度时按 DOUBLE 处理。
REAL REAL UNSIGNED REAL UNSIGNED ZEROFILL DOUBLE DOUBLE UNSIGNED DOUBLE UNSIGNED ZEROFILL DOUBLE PRECISION DOUBLE PRECISION UNSIGNED DOUBLE PRECISION UNSIGNED ZEROFILL REAL(p, s) DOUBLE(p, s)	DOUBLE
NUMERIC(p, s) NUMERIC(p, s) UNSIGNED NUMERIC(p, s) UNSIGNED ZEROFILL DECIMAL(p, s) DECIMAL(p, s) UNSIGNED DECIMAL(p, s) UNSIGNED ZEROFILL FIXED(p, s) FIXED(p, s) UNSIGNED FIXED(p, s) UNSIGNED ZEROFILL 其中 p <= 38	DECIMAL(p, s)
NUMERIC(p, s) 等 decimal 类型其中 38 < p <= 65	STRING	MySQL 的 decimal 精度最大为 65，而 Flink decimal 精度上限为 38，因此超过 38 时映射为 STRING 以避免精度丢失。
DATE	DATE
TIME [(p)]	TIME [(p)]
DATETIME [(p)]	TIMESTAMP [(p)]	未指定精度时按 TIMESTAMP(0) 处理。
TIMESTAMP [(p)]	TIMESTAMP_LTZ [(p)]	未指定精度时按 TIMESTAMP_LTZ(0) 处理。
CHAR(n)	CHAR(n)
VARCHAR(n)	VARCHAR(n)	VARCHAR(0) 映射为 STRING。
BINARY(n)	BINARY(n)
VARBINARY(n)	VARBINARY(n)
TINYTEXT TEXT MEDIUMTEXT LONGTEXT LONG LONG VARCHAR JSON ENUM	STRING	JSON 类型以 JSON 格式字符串的形式读出，格式受 use.legacy.json.format 影响。
TINYBLOB BLOB MEDIUMBLOB LONGBLOB	BYTES
SET	ARRAY(STRING)
GEOMETRY POINT LINESTRING POLYGON MULTIPOINT MULTILINESTRING MULTIPOLYGON GEOMETRYCOLLECTION	STRING	空间类型会转换为固定格式的 GeoJSON 字符串，详见「空间数据类型映射」章节。

Spatial data in MySQL	Json String converted in Flink
POINT(1 1)	{"coordinates":[1,1],"type":"Point","srid":0}
LINESTRING(3 0, 3 3, 3 5)	{"coordinates":[[3,0],[3,3],[3,5]],"type":"LineString","srid":0}
POLYGON((1 1, 2 1, 2 2, 1 2, 1 1))	{"coordinates":[[[1,1],[2,1],[2,2],[1,2],[1,1]]],"type":"Polygon","srid":0}
MULTIPOINT((1 1),(2 2))	{"coordinates":[[1,1],[2,2]],"type":"MultiPoint","srid":0}
MULTILINESTRING((1 1,2 2,3 3),(4 4,5 5))	{"coordinates":[[[1,1],[2,2],[3,3]],[[4,4],[5,5]]],"type":"MultiLineString","srid":0}
MULTIPOLYGON(((0 0, 10 0, 10 10, 0 10, 0 0)), ((5 5, 7 5, 7 7, 5 7, 5 5)))	{"coordinates":[[[[0,0],[10,0],[10,10],[0,10],[0,0]]],[[[5,5],[7,5],[7,7],[5,7],[5,5]]]],"type":"MultiPolygon","srid":0}
GEOMETRYCOLLECTION(POINT(10 10), POINT(30 30), LINESTRING(15 15, 20 20))	{"geometries":[{"type":"Point","coordinates":[10,10]},{"type":"Point","coordinates":[30,30]},{"type":"LineString","coordinates":[[15,15],[20,20]]}],"type":"GeometryCollection","srid":0}

CDC type	MySQL 目标字段类型	NOTE
BOOLEAN	TINYINT(1)
TINYINT	TINYINT
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
FLOAT	FLOAT
DOUBLE	DOUBLE
DECIMAL(p, s)	DECIMAL(p, s)
CHAR(n)	CHAR(n)，超长时降级为 VARCHAR / TEXT	n 不超过 255 时使用 CHAR(n)，超过时按 VARCHAR 的长度规则处理。
VARCHAR(n) / STRING	VARCHAR(n) / TEXT / MEDIUMTEXT / LONGTEXT	按字符长度自动选择：不超过 16383 用 VARCHAR(n)，不超过 65535 用 TEXT，不超过 4194303 用 MEDIUMTEXT，更长用 LONGTEXT，保证目标字段一定装得下。
BINARY(n)	BINARY(n) / VARBINARY(n) / LONGBLOB	n 不超过 255 用 BINARY(n)，不超过 65535 用 VARBINARY(n)，更长用 LONGBLOB。
VARBINARY(n) / BYTES	VARBINARY(n) / LONGBLOB	n 不超过 65535 用 VARBINARY(n)，更长用 LONGBLOB。
DATE	DATE
TIME(p)	TIME(p)	小数秒精度超过 6 时按 6 处理。
TIMESTAMP(p)	DATETIME(p)	不带时区的时间戳写入 DATETIME，小数秒精度超过 6 时按 6 处理。
TIMESTAMP_LTZ(p)	DATETIME(p)	先按 pipeline 的 local-time-zone 换算成挂钟时间再写入 DATETIME，因此落库结果不受 JVM 默认时区和服务端会话时区影响。
TIMESTAMP WITH TIME ZONE(p)	TIMESTAMP(p)	小数秒精度超过 6 时按 6 处理。
ARRAY MAP ROW	JSON	复合类型会序列化为 JSON 字符串后写入。ARRAY 内的二进制元素按 Base64 编码。

MySQL 连接器

本页目录：

前置准备

版本说明

使用范围

快速示例

MySQL 同步到 Doris

MySQL 同步到 MySQL

Source 配置项

常用配置项

高级配置

启动模式

Sink 配置项

Source 数据类型映射

空间数据类型映射

Sink 数据类型映射