流计算 Oceanus MySQL 连接器

一、前置准备

使用 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，多个作业之间以及与现有从库之间都不能重复。

二、版本说明

三、使用范围

可作为 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/Shanghaisink: type: doris name: Doris Sink fenodes: 127.0.0.1:8030 username: root password: passpipeline: 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-5404sink: 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: 5000pipeline: name: MySQL to MySQL Pipeline parallelism: 4

六、常用配置项

七、高级配置

下列参数用于性能调优和特殊场景，请结合实际场景使用。

八、启动模式

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 时的配置项如下：

Sink 侧的时区取自 pipeline 配置 local-time-zone，未配置时使用作业运行环境的系统时区。

十、Source 数据类型映射

MySQL 作为 Source 时，按下表把读取到的 MySQL 字段类型转换为 CDC 类型。

未在上表中出现的字段类型暂不支持，同步包含此类字段的表会导致作业失败。

十一、空间数据类型映射

除 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 是包含所有空间数据的数组。

各类空间数据的转换示例如下：

十二、Sink 数据类型映射

MySQL 作为 Sink 时，会自动为目标表生成 CREATE TABLE 语句（若表不存在），并按下列规则把 CDC 类型转换为目标库的 MySQL 字段类型。