MySQL YAML connector-Realtime Compute for Apache Flink(Flink)-阿里云帮助中心

The MySQL connector can be used as a data source in data ingestion YAML jobs.

Prerequisites

Before you use a MySQL CDC source table, you must complete the prerequisite operations described in Configure MySQL.

ApsaraDB RDS for MySQL

Perform a network probe to ensure network connectivity to Realtime Compute for Apache Flink.
MySQL version: 5.6, 5.7, 8.0.x, or 8.4.
Binary logging must be enabled. This is enabled by default.
The binary log format must be ROW. This is the default format.
The `binlog_row_image` parameter must be set to FULL. This is the default setting.
Binary Log Transaction Compression must be disabled. This feature was introduced in MySQL 8.0.20 and is disabled by default.
A MySQL user has been created with the SELECT, SHOW DATABASES, REPLICATION SLAVE, and REPLICATION CLIENT permissions.
Create a MySQL database and table. For more information, see Create a database and an account for an ApsaraDB RDS for MySQL instance. Use a privileged account to create the MySQL database to prevent operation failures due to insufficient permissions.
Configure an IP address whitelist. For more information, see Configure an IP address whitelist for an ApsaraDB RDS for MySQL instance.

PolarDB for MySQL

Perform a network probe to ensure network connectivity to Realtime Compute for Apache Flink.
MySQL version: 5.6, 5.7, 8.0.x, or 8.4.
Binary logging must be enabled. This is disabled by default.
The binary log format must be ROW. This is the default format.
The `binlog_row_image` parameter must be set to FULL. This is the default setting.
Binary Log Transaction Compression must be disabled. This feature was introduced in MySQL 8.0.20 and is disabled by default.
You have created a MySQL user with the SELECT, SHOW DATABASES, REPLICATION SLAVE, and REPLICATION CLIENT permissions.
Create a MySQL database and table. For more information, see Create a database and an account for a PolarDB for MySQL cluster. Use a privileged account to create the MySQL database to prevent operation failures due to insufficient permissions.
Configure an IP address whitelist. For more information, see Configure an IP address whitelist for a PolarDB for MySQL cluster.

Self-managed MySQL

Perform a network probe to ensure network connectivity to Realtime Compute for Apache Flink.
MySQL version: 5.6, 5.7, 8.0.x, or 8.4.
Binary logging must be enabled. This is disabled by default.
The binary log format must be ROW. The default format is STATEMENT.
The `binlog_row_image` parameter must be set to FULL. This is the default setting.
Binary Log Transaction Compression must be disabled. This feature was introduced in MySQL 8.0.20 and is disabled by default.
Create a MySQL user and grant the SELECT, SHOW DATABASES, REPLICATION SLAVE, and REPLICATION CLIENT permissions.
Create a MySQL database and table. For more information, see Create a database and an account for a self-managed MySQL instance. Use a privileged account to create the MySQL database to prevent operation failures due to insufficient permissions.
Configure an IP address whitelist. For more information, see Configure an IP address whitelist for a self-managed MySQL instance.

Limits

General limits

The MySQL CDC connector does not support the Binary Log Transaction Compression feature. Therefore, when you use the MySQL CDC connector to consume incremental data, ensure that Binary Log Transaction Compression is disabled. Otherwise, the connector may fail to retrieve incremental data.

ApsaraDB RDS for MySQL limits

For ApsaraDB RDS for MySQL, do not read data from a secondary database or read-only replica. This is because the default binary log retention period for secondary databases and read-only replicas is short. If binary logs expire and are cleared, the job may fail to consume the binary log data and report an error.
ApsaraDB RDS for MySQL enables parallel primary/secondary synchronization by default but does not guarantee a consistent transaction order between the primary and secondary instances. This may cause data to be missed during a primary/secondary switchover and checkpoint recovery. To avoid this issue, you can manually enable the `slave_preserve_commit_order` option for ApsaraDB RDS for MySQL.

PolarDB for MySQL limits

MySQL CDC source tables do not support reading data from Multi-master Cluster Architecture clusters of PolarDB for MySQL V1.0.19 and earlier. For more information, see What is a Multi-master Cluster?. The binary logs generated by these clusters may contain duplicate table IDs. This can cause schema mapping errors in the CDC source table, which leads to errors when parsing binary log data.

Open source MySQL limits

By default, MySQL maintains the transaction order during primary/secondary binary log replication. If a MySQL replica has parallel replication enabled (slave_parallel_workers > 1) but does not have slave_preserve_commit_order=ON enabled, its transaction commit order may be inconsistent with the primary database. When Flink CDC recovers from a checkpoint, it may miss data because of the disordered sequence. You can set `slave_preserve_commit_order` = ON on the MySQL replica. Alternatively, you can set `slave_parallel_workers` = 1, but this will sacrifice replication performance.

Usage notes

Set a server ID to avoid binary log consumption conflicts.
During the full data reading phase, you cannot save a savepoint, add a new table to or delete a table from the source table, and then restart the job from the savepoint. This will cause the job to fail to read data.

Data ingestion

You can use the MySQL connector as a data source in a data ingestion YAML job.

Syntax

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

Configuration items

Parameter	Description	Required	Data type	Default value	Notes
type	The data source type.	Yes	STRING	None	The value must be mysql.
name	The data source name.	No	STRING	None	None.
hostname	The IP address or hostname of the MySQL database.	Yes	STRING	None	We recommend that you specify a VPC address. Note If the MySQL database and Realtime Compute for Apache Flink are not in the same VPC, you must establish a cross-VPC network connection or use a public endpoint to access the database. For more information, see Manage and operate workspaces and How can a fully managed Flink cluster access the Internet?.
username	The username for the MySQL database service.	Yes	STRING	None	None.
password	The password for the MySQL database service.	Yes	STRING	None	None.
tables	The MySQL data tables to be synchronized.	Yes	STRING	None	This parameter supports regular expressions to read data from multiple tables. You can use commas to separate multiple regular expressions. Note Do not use the start-of-string `^` and end-of-string `$` matching characters in the regular expression. In VVR 11.2, the period is used to split the regular expression to get the database part. Start and end matching characters will make the resulting database regular expression unusable. For example, you must change `^db.user_[0-9]+$` to `db.user_[0-9]+`. The period is used to separate the database name and table name. To use a period to match any character, you must escape it with a backslash. For example: db0.\., db1.user_table_[0-9]+, db[1-2].[app\|web]order_\..
tables.exclude	The tables to be excluded from synchronization.	No	STRING	None	This parameter supports regular expressions to exclude multiple tables. You can use commas to separate multiple regular expressions. Note The period is used to separate the database name and table name. To use a period to match any character, you must escape it with a backslash. For example: db0.\., db1.user_table_[0-9]+, db[1-2].[app\|web]order_\..
port	The port number of the MySQL database service.	No	INTEGER	3306	None.
schema-change.enabled	Specifies whether to send schema change events.	No	BOOLEAN	true	None.
server-id	A numeric ID or range for the database client used for synchronization.	No	STRING	A random value between 5400 and 6400 is generated.	This ID must be globally unique within the MySQL cluster. Set a different ID for each job that connects to the same database. This parameter also supports an ID range format, such as 5400-5408. Note When incremental reading is enabled, concurrent reading is supported. In this case, we recommend that you set an ID range so that each concurrent reader uses a different ID.
jdbc.properties.*	Custom connection parameters in the JDBC URL.	No	STRING	None	You can pass custom connection parameters. For example, to not use the SSL protocol, you can configure 'jdbc.properties.useSSL' = 'false'. For more information about the supported connection parameters, see MySQL Configuration Properties.
debezium.*	Custom parameters for Debezium to read binary logs.	No	STRING	None	You can pass custom Debezium parameters. For example, use 'debezium.event.deserialization.failure.handling.mode'='ignore' to specify the handling logic for parsing errors. Warning Do not modify Debezium parameters arbitrarily. This may cause the connector to read data incorrectly. For example, the debezium.binlog.buffer.size parameter is not allowed to be configured.
scan.incremental.snapshot.chunk.size	The size of each chunk in number of rows.	No	INTEGER	8096	MySQL tables are split into multiple chunks for reading. The data of a chunk is cached in memory before it is fully read. The fewer rows each chunk contains, the greater the total number of chunks in the table. Although this reduces the granularity of fault recovery, it may lead to OOM errors and lower overall throughput. Therefore, you need to make a trade-off and set a reasonable chunk size.
scan.snapshot.fetch.size	The maximum number of records to pull at a time when reading the full data of a table.	No	INTEGER	1024	None.
scan.startup.mode	The startup mode for data consumption.	No	STRING	initial	Valid values: initial (default): On the first startup or a stateless startup, the connector scans the full historical data and then reads the latest binary log data. latest-offset: On the first startup or a stateless startup, the connector does not scan the historical data. It starts reading from the end of the binary log, which means it only reads the latest changes made after the connector starts. earliest-offset: The connector does not scan the historical data. It starts reading from the earliest available binary log. specific-offset: The connector does not scan the historical data. It starts from a specific binary log offset. You can specify the offset by configuring both scan.startup.specific-offset.file and scan.startup.specific-offset.pos, or by configuring only scan.startup.specific-offset.gtid-set to start from a specific GTID set. timestamp: The connector does not scan the historical data. It starts reading the binary log from a specified timestamp. The timestamp is specified by scan.startup.timestamp-millis in milliseconds. Important For the earliest-offset, specific-offset, and timestamp startup modes, if the table schema at the startup time is different from the schema at the specified start offset time, the job will report an error due to the schema mismatch. In other words, when using these three startup modes, you must ensure that the schema of the corresponding table does not change between the specified binary log consumption position and the job startup time.
scan.startup.specific-offset.file	The binary log filename for the start offset when using the specific-offset startup mode.	No	STRING	None	When you use this parameter, you must set scan.startup.mode to specific-offset. Example filename format: `mysql-bin.000003`.
scan.startup.specific-offset.pos	The offset within the specified binary log file for the start offset when using the specific-offset startup mode.	No	INTEGER	None	When you use this parameter, you must set scan.startup.mode to specific-offset.
scan.startup.specific-offset.gtid-set	The GTID set for the start offset when using the specific-offset startup mode.	No	STRING	None	When you use this parameter, you must set scan.startup.mode to specific-offset. Example GTID set format: `24DA167-0C0C-11E8-8442-00059A3C7B00:1-19`.
scan.startup.timestamp-millis	The timestamp in milliseconds for the start offset when using the timestamp startup mode.	No	LONG	None	When you use this parameter, you must set scan.startup.mode to timestamp. The timestamp unit is milliseconds. Important When you specify a time, MySQL CDC attempts to read the initial event of each binary log file to determine its timestamp. It then locates the binary log file corresponding to the specified time. Make sure that the binary log file corresponding to the specified timestamp has not been cleared from the database and can be read.
server-time-zone	The session time zone used by the database.	No	STRING	If you do not specify this parameter, the system uses the environment time zone of the Flink job runtime as the database server time zone. This is the time zone of the zone you selected.	Example: Asia/Shanghai. This parameter controls how the TIMESTAMP type in MySQL is converted to the STRING type. For more information, see Debezium temporal values.
scan.startup.specific-offset.skip-events	The number of binary log events to skip when reading from a specified offset.	No	INTEGER	None	When you use this parameter, you must set scan.startup.mode to specific-offset.
scan.startup.specific-offset.skip-rows	The number of row changes to skip when reading from a specified offset. A single binary log event may correspond to multiple row changes.	No	INTEGER	None	When you use this parameter, you must set scan.startup.mode to specific-offset.
connect.timeout	The maximum time to wait for a connection to the MySQL database server to time out before retrying.	No	DURATION	30 s	None.
connect.max-retries	The maximum number of retries after a failed connection to the MySQL database service.	No	INTEGER	3	None.
connection.pool.size	The size of the database connection pool.	No	INTEGER	20	The database connection pool is used to reuse connections, which can reduce the number of database connections.
heartbeat.interval	The interval at which the source advances the binary log offset using heartbeat events.	No	DURATION	30s	Heartbeat events are used to advance the binary log offset in the source. This is very useful for tables in MySQL that are updated infrequently. For such tables, the binary log offset cannot advance automatically. Heartbeat events can push the binary log offset forward, which prevents issues caused by an expired binary log offset. An expired binary log offset can cause the job to fail and be unrecoverable, requiring a stateless restart.
rds.region-id	The region ID of the Alibaba Cloud ApsaraDB RDS for MySQL instance.	Required when using the feature to read archived logs from OSS.	STRING	None	For more information about region IDs, see Regions and zones. Important Because the GTID string for MySQL CDC is randomly generated and not monotonically increasing like binary log file offsets, locating a GTID in a file requires downloading and parsing all archived logs from OSS. This process is very resource-intensive and time-consuming, making features that rely on GTID offsets infeasible. Therefore, the OSS archived log feature only supports starting from a specified timestamp or a specified binary log file offset. It does not support starting from a specified GTID, nor does it support scenarios with primary/secondary switchovers in the archived logs, because MySQL primary/secondary switchovers rely on GTIDs. Evaluate this feature carefully before use.
rds.access-key-id	The AccessKey ID of the Alibaba Cloud ApsaraDB RDS for MySQL account.	Required when using the feature to read archived logs from OSS.	STRING	None	For more information, see How do I view the AccessKey ID and AccessKey secret? Important To prevent your AccessKey information from being leaked, use the secret management feature to specify the AccessKey ID. For more information, see Manage variables.
rds.access-key-secret	The AccessKey secret of the Alibaba Cloud ApsaraDB RDS for MySQL account.	Required when using the feature to read archived logs from OSS.	STRING	None	For more information, see How do I view the AccessKey ID and AccessKey secret? Important To prevent your AccessKey information from being leaked, use the secret management feature to specify the AccessKey secret. For more information, see Manage variables.
rds.db-instance-id	The ID of the Alibaba Cloud ApsaraDB RDS for MySQL instance.	Required when using the feature to read archived logs from OSS.	STRING	None	None.
rds.main-db-id	The primary database number of the Alibaba Cloud ApsaraDB RDS for MySQL instance.	No	STRING	None	For more information about how to obtain the primary database number, see ApsaraDB RDS for MySQL log backup. Note If this parameter is not specified, VVR 11.7 and later automatically query the primary database number based on the ApsaraDB RDS for MySQL connection information.
rds.download.timeout	The timeout period for downloading a single archived log from OSS.	No	DURATION	60s	None.
rds.endpoint	The service endpoint for obtaining OSS binary log information.	No	STRING	None	For more information about the valid values, see Endpoints.
rds.binlog-directory-prefix	The directory prefix for storing binary log files.	No	STRING	rds-binlog-	None.
rds.use-intranet-link	Specifies whether to use the internal network to download binary log files.	No	BOOLEAN	true	None.
rds.binlog-directories-parent-path	The absolute path of the parent directory for storing binary log files.	No	STRING	None	None.
chunk-meta.group.size	The size of the chunk metadata.	No	INTEGER	1000	If the metadata is larger than this value, it is split into multiple parts for transmission.
chunk-key.even-distribution.factor.lower-bound	The lower bound of the chunk distribution factor for even sharding.	No	DOUBLE	0.05	If the distribution factor is less than this value, uneven sharding is used. Chunk distribution factor = (MAX(chunk-key) - MIN(chunk-key) + 1) / Total number of data rows.
chunk-key.even-distribution.factor.upper-bound	The upper bound of the chunk distribution factor for even sharding.	No	DOUBLE	1000.0	If the distribution factor is greater than this value, uneven sharding is used. Chunk distribution factor = (MAX(chunk-key) - MIN(chunk-key) + 1) / Total number of data rows.
scan.incremental.close-idle-reader.enabled	Specifies whether to close idle readers after the snapshot is complete.	No	BOOLEAN	false	For this configuration to take effect, you must set `execution.checkpointing.checkpoints-after-tasks-finish.enabled` to true.
scan.only.deserialize.captured.tables.changelog.enabled	In the incremental phase, specifies whether to deserialize only the change events of the specified tables.	No	BOOLEAN	The default value is false in VVR 8.x versions. The default value is true in VVR 11.1 and later.	Valid values: true: Deserializes only the change data of the target tables to accelerate binary log reading. false (default): Deserializes the change data of all tables.
scan.parallel-deserialize-changelog.enabled	In the incremental phase, specifies whether to use multiple threads to parse change events.	No	BOOLEAN	false	Valid values: true: Uses multiple threads in the change event deserialization phase while maintaining the order of binary log events to accelerate reading. false (default): Uses a single thread in the event deserialization phase. Note Supported only in VVR 8.0.11 and later.
scan.parallel-deserialize-changelog.handler.size	The number of event handlers when using multiple threads to parse change events.	No	INTEGER	2	Note Supported only in VVR 8.0.11 and later.
metadata-column.include-list	The metadata columns to be passed to the downstream.	No	STRING	None	The available metadata includes `op_ts`, `es_ts`, `query_log`, `file`, and `pos`. You can use commas to separate multiple metadata columns. Note The MySQL CDC YAML connector does not require or support adding database name, table name, and `op_type` metadata columns. You can directly use `__data_event_type__` in a Transform expression to get the change data type, or use `__schema_name__` and `__table_name__` to get the database name and table name. Important The `file` metadata column represents the binary log file where the data is located. It is "" during the full phase and the binary log filename during the incremental phase. The `pos` metadata column represents the offset of the data in the binary log file. It is "0" during the full phase and the data offset in the binary log file during the incremental phase. These two metadata columns are supported starting from VVR 11.5. The `es_ts` metadata column represents the start time of the corresponding transaction for the changelog on MySQL. It is supported only for MySQL 8.0.x. Do not add this metadata column when using earlier versions of MySQL. The `op_ts` timestamp is accurate to the second, while the `es_ts` timestamp is accurate to the millisecond.
scan.newly-added-table.enabled	When restarting from a checkpoint, specifies whether to synchronize newly added tables that were not matched during the previous startup or to remove tables from the state that are no longer matched.	No	BOOLEAN	false	This takes effect when restarting from a checkpoint or savepoint. Important During the full data reading phase, you cannot save a savepoint, add a new table to or delete a table from the source table, and then restart the job from the savepoint. This will cause the job to fail to read data.
scan.binlog.newly-added-table.enabled	In the incremental phase, specifies whether to send data from newly added tables that are matched.	No	BOOLEAN	false	Cannot be enabled at the same time as `scan.newly-added-table.enabled`.
scan.incremental.snapshot.chunk.key-column	Specifies a column for certain tables to be used as the splitting column for sharding during the snapshot phase.	No	STRING	None	Use a colon `:` to connect the table name and column name to define a rule. The table name can be a regular expression. You can define multiple rules by separating them with a semicolon `;`. For example: `db1.user_table_[0-9]+:col1;db[1-2].[app\|web]_order_\\.*:col2`. Required for tables without a primary key. The selected column must be of a non-null type (NOT NULL). Optional for tables with a primary key. Only one column can be selected from the primary key.
scan.parse.online.schema.changes.enabled	In the incremental phase, specifies whether to attempt to parse RDS lockless change DDL events.	No	BOOLEAN	false	Valid values: true: Parses RDS lockless change DDL events. false (default): Does not parse RDS lockless change DDL events. This is an experimental feature. Before performing an online lockless change, take a snapshot of the Flink job for recovery. Note Supported only in VVR 11.0 and later.
scan.incremental.snapshot.backfill.skip	Specifies whether to skip backfill during the snapshot reading phase.	No	BOOLEAN	false	Valid values: true: Skips backfill during the snapshot reading phase. false (default): Does not skip backfill during the snapshot reading phase. Backfill applies only during the snapshot query of a single chunk and does not cover the entire full-read phase. When backfill is skipped, each chunk's snapshot query reads the latest table data at that instant; updates that occur on a chunk after it has been read are not merged during the full-read phase and are read from the Binlog after entering the incremental phase. For example, an update to chunk5 that occurs while chunk5 is being snapshotted is reflected directly in chunk5's snapshot; if chunk5 is updated after the reader has advanced to chunk80, the update is applied later from the Binlog during the incremental phase. Important When enabled, changes that occur during or after a chunk's scan are still delivered from the Binlog in the incremental phase and may be duplicated. Only at-least-once semantics are guaranteed. Enable this only when the downstream sink supports idempotent writes by primary key. Note Supported only in VVR 11.1 and later.
treat-tinyint1-as-boolean.enabled	Specifies whether to treat the TINYINT(1) type as a Boolean type.	No	BOOLEAN	true	Valid values: true (default): Treats the TINYINT(1) type as a Boolean type. false: Does not treat the TINYINT(1) type as a Boolean type.
treat-timestamp-as-datetime-enabled	Specifies whether to treat the TIMESTAMP type as a DATETIME type.	No	BOOLEAN	false	Valid values: true: Treats the MySQL TIMESTAMP type as a DATETIME type and maps it to the CDC TIMESTAMP type. false (default): Maps the MySQL TIMESTAMP type to the CDC TIMESTAMP_LTZ type. The MySQL TIMESTAMP type stores UTC time and is affected by the time zone. The MySQL DATETIME type stores literal time and is not affected by the time zone. When enabled, this parameter converts MySQL TIMESTAMP type data to DATETIME type based on the server-time-zone.
include-comments.enabled	Specifies whether to synchronize table and column comments.	No	BOOELEAN	false	Valid values: true: Synchronizes table and column comments. false (default): Does not synchronize table and column comments. Enabling this option increases the memory usage of the job.
scan.incremental.snapshot.unbounded-chunk-first.enabled	Specifies whether to dispatch unbounded chunks first during the snapshot reading phase.	No	BOOELEAN	false	Valid values: true: Dispatches unbounded chunks first during the snapshot reading phase. false (default): Does not dispatch unbounded chunks first during the snapshot reading phase. This is an experimental feature. Enabling it can reduce the risk of OOM errors on the TaskManager when synchronizing the last chunk during the snapshot phase. Add this parameter before the first startup of the job. Note Supported only in VVR 11.1 and later.
binlog.session.network.timeout	The network timeout for the binary log connection.	No	DURATION	10m	If set to 0s, the default timeout of the MySQL server is used. Note Supported only in VVR 11.5 and later.
scan.rate-limit.records-per-second	Limits the maximum number of records sent by the source per second.	No	LONG	None	This is applicable to scenarios where data reading needs to be limited. This limit is effective in both the full and incremental phases. The `numRecordsOutPerSecond` metric of the source reflects the number of records output by the entire data stream per second. You can adjust this parameter based on this metric. In the full data reading phase, you usually need to reduce the number of rows read in each batch. You can reduce the value of the `scan.incremental.snapshot.chunk.size` parameter. Note Supported only in VVR 11.5 and later.
include-binlog-meta.enable	Specifies whether to include the original MySQL binary log information, such as GTID and binary log offset, in the message.	No	Boolean	false	This is applicable to original binary log synchronization scenarios, such as replacing an existing Canal synchronization link. Note Supported only in VVR 11.6 and later.
scan.binlog.tolerate.gtid-holes	Enabling this parameter ignores gaps in the GTID sequence, allowing the job to bypass discontinuous events and continue running.	No	Boolean	false	Before enabling this parameter, you must ensure that the job's start offset has not expired. If the job starts from a cleared or expired GTID offset, the engine will silently skip the missing logs, which will lead to data loss. Note This parameter is supported only in VVR 11.6 and later.
scan.emit.create-table-events.in-batch.enabled	Specifies whether to batch send table schemas during the job initialization phase.	No	Boolean	false	This is an experimental feature. Enable this option when a single job synchronizes many tables. Note This parameter is supported only in VVR 11.4 and later.

Reuse an existing catalog

Starting from VVR 11.5, you can directly reference a built-in MySQL catalog created on the Data Management page in a Flink CDC data ingestion job. This reduces the manual effort of writing connection properties.

source:
  type: mysql
  using.built-in-catalog: mysql_rds_catalog

Currently, data ingestion jobs support the automatic reuse of the following MySQL catalog parameters:

hostname
port
username
password
catalog.table.metadata-columns
catalog.table.treat-tinyint1-as-boolean

If you want to override any of these automatically reused parameters, you can explicitly write the corresponding YAML parameter. The explicitly written parameter has a higher priority.

Type mapping

The following table shows the data type mapping for data ingestion.

MySQL CDC field type	CDC field type
TINYINT(n)	TINYINT
SMALLINT	SMALLINT
TINYINT UNSIGNED
TINYINT UNSIGNED ZEROFILL
YEAR
INT	INT
MEDIUMINT
MEDIUMINT UNSIGNED
MEDIUMINT UNSIGNED ZEROFILL
SMALLINT UNSIGNED
SMALLINT UNSIGNED ZEROFILL
BIGINT	BIGINT
INT UNSIGNED
INT 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] and p <= 38	DECIMAL(p, s)
DECIMAL(p, s) [UNSIGNED] [ZEROFILL] and p <= 38
FIXED(p, s) [UNSIGNED] [ZEROFILL] and p <= 38
BOOLEAN	BOOLEAN
BIT(1)
TINYINT(1)
DATE	DATE
TIME [(p)]	TIME [(p)]
DATETIME [(p)]	TIMESTAMP [(p)]
TIMESTAMP [(p)]	The mapping depends on the value of the `treat-timestamp-as-datetime-enabled` parameter: `true`:TIMESTAMP[(p)] `false`:TIMESTAMP_LTZ[(p)]
CHAR(n)	CHAR(n)
VARCHAR(n)	VARCHAR(n)
BIT(n)	BINARY(⌈(n + 7) / 8⌉)
BINARY(n)	BINARY(n)
VARBINARY(N)	VARBINARY(N)
NUMERIC(p, s) [UNSIGNED] [ZEROFILL] and 38 < p <= 65	STRING Note In MySQL, the decimal data type has a precision of up to 65, but in Flink, the precision is limited to 38. Therefore, if you define a decimal column with a precision greater than 38, you should map it to a string to avoid loss of precision.
DECIMAL(p, s) [UNSIGNED] [ZEROFILL] and 38 < p <= 65
FIXED(p, s) [UNSIGNED] [ZEROFILL] and 38 < p <= 65
TINYTEXT	STRING
TEXT
MEDIUMTEXT
LONGTEXT
ENUM
JSON	STRING Note The JSON data type is converted to a JSON-formatted string in Flink.
GEOMETRY	STRING Note Spatial data types in MySQL are converted to strings with a fixed JSON format. For more information, see MySQL Spatial Data Type Mapping.
POINT
LINESTRING
POLYGON
MULTIPOINT
MULTILINESTRING
MULTIPOLYGON
GEOMETRYCOLLECTION
TINYBLOB	BYTES Note For the BLOB data type in MySQL, only blobs with a length no greater than 2,147,483,647 (2**31-1) are supported.
BLOB
MEDIUMBLOB
LONGBLOB

Set a server ID to avoid binary log consumption conflicts

When a data ingestion job reads binary logs, the source registers with MySQL as a replication client by using a server-id. If multiple jobs or other replication clients use the same server ID, binary log consumption conflicts occur and the jobs fail. Note the following when you configure server IDs:

By default, a server-id is a random single value between 5400 and 6400. If multiple jobs use default values, conflicts may occur. We recommend that you explicitly configure non-overlapping IDs for the jobs.
If the source parallelism is greater than 1, you must configure a server ID range. The number of available IDs in the range must be no less than the parallelism, and each parallel reader uses a different ID.

Example: If the source parallelism is 4, configure a range that contains four IDs.

source:
  type: mysql
  name: MySQL Source
  hostname: <hostname>
  port: 3306
  username: <username>
  password: <password>
  tables: app_db.\.*
  server-id: 5400-5403

sink:
  type: hologres

If multiple jobs read from the same MySQL instance, assign a non-overlapping range to each job. For example, job A uses 5400-5403, and job B uses 5404-5407.

Accelerate binary log reading

When you use the MySQL connector as a data ingestion data source, it parses binary log files to generate various change messages during the incremental phase. The binary log files record all table changes in binary format. You can accelerate the parsing of binary log files in the following ways.

Enable parallel parsing and parsing filters (This feature requires Realtime Compute for Apache Flink with Ververica Runtime (VVR) 8.0.7 or later. It is not available in the community edition of the MySQL CDC connector.)
- Enable the scan.only.deserialize.captured.tables.changelog.enabled option to parse change events only for specified tables.
- Enable the scan.parallel-deserialize-changelog.enabled option to use multiple threads to parse the binary log file and deliver events to the consumer queue in order. When you enable this option, you usually need to increase the TaskManager CPU as well.
Optimize Debezium parameters
```
debezium.max.queue.size: 162580
debezium.max.batch.size: 40960
debezium.poll.interval.ms: 50
```
- debezium.max.queue.size: The maximum number of records that the blocking queue can hold. When Debezium reads an event stream from the database, it places the events in a blocking queue before writing them downstream. The default value is 8192.
- debezium.max.batch.size: The maximum number of events that the connector processes in each iteration. The default value is 2048.
- debezium.poll.interval.ms: The number of milliseconds the connector should wait before it requests new change events. The default value is 1000 milliseconds, or 1 second.

Usage example:

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: ${mysql.source.table}
  server-id: 7601-7604
  # Debezium configuration
  debezium.max.queue.size: 162580
  debezium.max.batch.size: 40960
  debezium.poll.interval.ms: 50
  # Enable parsing filter
  scan.only.deserialize.captured.tables.changelog.enabled: true

The binary log consumption capacity of the MySQL CDC Enterprise Edition is 85 MB/s, which is about twice that of the open source community version. When the generation speed of binary log files exceeds 85 MB/s (that is, one 512 MB file every 6 seconds), the latency of the Flink job continues to increase. The processing latency gradually decreases after the binary log file generation speed slows down. If a binary log file contains a large transaction, the processing latency may temporarily increase. The latency decreases after the log for that transaction is read.

Diagnose data latency to optimize job throughput

If you experience data latency during the incremental phase, analyze the issue by following these steps:

Check the currentFetchEventTimeLag and currentEmitEventTimeLag metrics on the Overview page. The currentFetchEventTimeLag metric represents the latency in reading data from the binary log. The currentEmitEventTimeLag metric represents the latency in reading data for the tables relevant to the job from the binary log.

Scenario	Description
`currentFetchEventTimeLag` is low, while `currentEmitEventTimeLag` is high and rarely updates.	A low `currentFetchEventTimeLag` indicates that pulling the binary log from the database is efficient. However, the binary log contains little data for the tables that the job needs to read. Therefore, `currentEmitEventTimeLag` rarely updates. This is expected behavior.
Both `currentFetchEventTimeLag` and `currentEmitEventTimeLag` are high.	This indicates that the source table has poor read performance. You can proceed to the subsequent steps in this section for optimization.

Backpressure can reduce the rate at which the source sends data to downstream operators. You may observe that sourceIdleTime periodically increases, and both currentFetchEventTimeLag and currentEmitEventTimeLag continuously grow. To resolve this, increase the parallelism of the node where the backpressure originates.
Check the TM CPU Usage metric on the CPU page and the TM GC Time metric on the JVM page to determine if there are insufficient CPU or memory resources. You can increase job resources to optimize read performance.

Read archived binlogs from OSS

When you use an ApsaraDB RDS for MySQL instance as a data source, you can read log backups stored in OSS. If the file corresponding to the specified timestamp or binary log position is stored in OSS, Flink automatically pulls the log file from OSS to the cluster. If the file is stored locally in the database, Flink automatically switches to reading through a database connection. This feature is available only in Realtime Compute for Apache Flink and is not supported in the community edition of the MySQL CDC connector.

To enable reading from OSS log backups, you must configure the ApsaraDB RDS for MySQL connection parameters. Example:

source:
  type: mysql
  hostname: <yourHostname>
  port: 3306
  username: <yourUsername>
  password: <yourPassword>
  tables: <yourTables>
  # RDS connection parameters for reading archived binlogs from OSS
  rds.region-id: cn-beijing
  rds.access-key-id: your_access_key_id
  rds.access-key-secret: your_access_key_secret
  rds.db-instance-id: rm-xxxxxxxx # Database instance ID.
  rds.main-db-id: 12345678 # Primary database ID.
  rds.endpoint: rds.aliyuncs.com

Data ingestion for database and schema synchronization

For jobs that only involve data synchronization logic, we recommend running them as data ingestion jobs. Data ingestion jobs are deeply optimized for data integration scenarios. For usage instructions, see Get Started with Flink CDC Data Ingestion and Develop a Flink CDC Data Ingestion Job (Public Preview).

The following code shows an example data ingestion job that synchronizes the entire app_db database from MySQL to Hologres, including subsequent schema changes from the upstream database:

source:
  type: mysql
  hostname: <hostname>
  port: 3306
  username: ${secret_values.mysqlusername}
  password: ${secret_values.mysqlpassword}
  tables: app_db.\.*
  server-id: 5400-5404

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <endpoint>
  dbname: <database-name>
  username: ${secret_values.holousername}
  password: ${secret_values.holopassword}

pipeline:
  name: Sync MySQL Database to Hologres

New table discovery in data ingestion

The MySQL data ingestion connector provides configuration options to support the discovery of new tables in two different scenarios.

Parameter	Description	Notes
`scan.newly-added-table.enabled`	When a job restarts from a checkpoint, this option synchronizes tables that were not discovered during the previous startup. It reads both snapshot and incremental data from these new tables.	This option is only supported when `scan.startup.mode` is set to `initial`. It has no effect in other startup modes.
`scan.binlog.newly-added-table.enabled`	During the incremental phase, this option automatically synchronizes data from newly discovered tables.	We recommend enabling this option when you first start the job. The job automatically parses `CREATE TABLE` DDL statements and synchronizes the data downstream. If you enable this option and restart the job after the database table has already been created, it can lead to incomplete data. In `initial` startup mode, DDL operations are not synchronized downstream until the snapshot phase is complete. Tables created during the snapshot phase cannot be automatically synchronized even if `scan.binlog.newly-added-table.enabled` is enabled.

Important

During the full-read phase, saving a savepoint and then adding or dropping source tables before restarting from the savepoint is not supported. Doing so prevents the job from reading data correctly.
Do not enable scan.newly-added-table.enabled and scan.binlog.newly-added-table.enabled simultaneously. Enabling both can cause data duplication.