This documentation is for an unreleased version of Apache Flink CDC. We recommend you use the latest stable version.

MySQL Connector #

MySQL CDC Pipeline 连接器允许从 MySQL 数据库读取快照数据和增量数据，并提供端到端的整库数据同步能力。 本文描述了如何设置 MySQL CDC Pipeline 连接器。

依赖配置 #

由于 MySQL Connector 采用的 GPLv2 协议与 Flink CDC 项目不兼容，我们无法在 jar 包中提供 MySQL 连接器。 您可能需要手动配置以下依赖，并在提交 YAML 作业时使用 Flink CDC CLI 的 --jar 参数将其传入：

示例 #

从 MySQL 读取数据同步到 Doris 的 Pipeline 可以定义如下：

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

sink:
  type: doris
  name: Doris Sink
  fenodes: 127.0.0.1:8030
  username: root
  password: pass

pipeline:
   name: MySQL to Doris Pipeline
   parallelism: 4

连接器配置项 #

启动模式 #

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

• initial （默认）：在第一次启动时对受监视的数据库表执行初始快照，并继续读取最新的 binlog。
• earliest-offset：跳过快照阶段，从可读取的最早 binlog 位点开始读取
• latest-offset：首次启动时，从不对受监视的数据库表执行快照， 连接器仅从 binlog 的结尾处开始读取，这意味着连接器只能读取在连接器启动之后的数据更改。
• specific-offset：跳过快照阶段，从指定的 binlog 位点开始读取。位点可通过 binlog 文件名和位置指定，或者在 GTID 在集群上启用时通过 GTID 集合指定。
• timestamp：跳过快照阶段，从指定的时间戳开始读取 binlog 事件。
• snapshot: 只进行快照阶段，跳过增量阶段，快照阶段读取结束后退出。

例如，可以在 YAML 配置文件中这样指定启动模式：

source:
  type: mysql
  scan.startup.mode: earliest-offset                    # Start from earliest offset
  scan.startup.mode: latest-offset                      # Start from latest offset
  scan.startup.mode: specific-offset                    # Start from specific offset
  scan.startup.mode: timestamp                          # Start from timestamp
  scan.startup.mode: snapshot                          # Read snapshot only
  scan.startup.specific-offset.file: 'mysql-bin.000003' # Binlog filename under specific offset startup mode
  scan.startup.specific-offset.pos: 4                   # Binlog position under specific offset mode
  scan.startup.specific-offset.gtid-set: 24DA167-...    # GTID set under specific offset startup mode
  scan.startup.timestamp-millis: 1667232000000          # Timestamp under timestamp startup mode
  # ...

可用的指标 #

指标系统能够帮助了解分片分发的进展， 下面列举出了支持的 Flink 指标 Flink metrics:

注意:

1. Group 名称是 namespace.schema.table，这里的 namespace 是实际的数据库名称， schema 是实际的 schema 名称， table 是实际的表名称。
2. 对于 MySQL，这里的 namespace 会被设置成默认值 “"，也就是一个空字符串，Group 名称的格式会类似于 test_database.test_table。

数据类型映射 #

空间数据类型映射 #

MySQL中除GEOMETRYCOLLECTION之外的空间数据类型都会转换为 Json 字符串，格式固定，如：

{"srid": 0 , "type": "xxx", "coordinates": [0, 0]}

字段srid标识定义几何体的 SRS，如果未指定 SRID，则 SRID 0 是新几何体值的默认值。 由于 MySQL 8+ 在定义空间数据类型时只支持特定的 SRID，因此在版本较低的MySQL中，字段srid将始终为 0。

字段type标识空间数据类型，例如POINT/LINESTRING/POLYGON。

字段coordinates表示空间数据的坐标。

对于GEOMETRYCOLLECTION，它将转换为 Json 字符串，格式固定，如：

{"srid": 0 , "type": "GeometryCollection", "geometries": [{"type":"Point","coordinates":[10,10]}]}

Geometrics字段是一个包含所有空间数据的数组。

不同空间数据类型映射的示例如下：

Back to top