MySQL

SQL

Use the MySQL connector in SQL jobs as a source, dimension, or sink table.

Syntax

CREATE TEMPORARY 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 parameters

• General

  Parameter	Description	Required	Type	Default	Notes
  connector	The connector type.	Yes	STRING	-	For a source table, the value can be mysql-cdc or mysql. These two values are equivalent. For a dimension table or sink table, the value must be mysql.
  hostname	The IP address or hostname of the MySQL database.	Yes	STRING	-	A VPC address is recommended. Note If your MySQL database and Realtime Compute for Apache Flink workspace are in different VPCs, establish a cross-VPC network connection or connect over the public network. Space Management and Operations and How can a fully managed Flink cluster access the public network?.
  username	The username for the MySQL database.	Yes	STRING	-	-
  password	The password for the MySQL database.	Yes	STRING	-	-
  database-name	The database name.	Yes	STRING	-	When used as a source, you can specify a regular expression to read data from multiple databases. When using regular expressions, avoid the ^ and $ symbols. See the notes for table-name.
  table-name	The table name.	Yes	STRING	-	When used as a source, you can specify a regular expression to read data from multiple tables. To improve performance when reading from multiple MySQL tables, submit multiple CTAS statements as a single job to avoid multiple binlog listeners. Run multiple CTAS statements in a single job. When you use regular expressions, avoid using the ^ and $ symbols to match the start and end. See the following description for the specific reasons. Note When a MySQL CDC source table matches table names by using a regular expression, it concatenates the database-name and table-name that you enter with the string \\. (for VVR versions before 8.0.1, the character . is used) to form a full-path regular expression, and then uses this expression to match the fully qualified names of tables in the MySQL database. For example, when you configure 'database-name'='db_.*' and 'table-name'='tb_.+', the connector will use the regular expression db_.*\\.tb_.+ (or db_.*.tb_.+ for versions earlier than 8.0.1) to match fully qualified table names to determine which tables to read.
  port	The port of the MySQL database.	No	INTEGER	3306	-

• Source-specific

  Parameter	Description	Required	Type	Default	Remarks
  server-id	A numeric ID for the database client.	No	STRING	Random integer between 5400 and 6400.	The server ID must be globally unique within the MySQL cluster. Assign a different ID for each job connecting to the same database. Also supports an ID range (e.g. 5400-5408). When incremental snapshot is enabled with concurrent readers, specify an ID range with at least as many IDs as the source parallelism. Server ID Usage.
  scan.incremental.snapshot.enabled	Enables or disables incremental snapshot.	No	BOOLEAN	true	Incremental snapshot is enabled by default. Compared to the legacy snapshot mechanism, it offers: Parallel snapshot reads. Chunk-level checkpointing during the snapshot phase. No global read lock (FLUSH TABLES WITH READ LOCK) required. To enable parallel reading, each concurrent reader requires a unique server ID. Therefore, the server-id must be set to a range, such as 5400-6400, and the size of the range must be greater than or equal to the source parallelism. Note This parameter is removed in Flink engine VVR 11.1 and later.
  scan.incremental.snapshot.chunk.size	The number of rows in each chunk.	No	INTEGER	8096	With incremental snapshot enabled, the table is split into multiple chunks. Each chunk is buffered in memory until fully read. Smaller chunks enable more granular fault recovery but can cause OOM issues and lower throughput.
  scan.snapshot.fetch.size	The maximum number of rows to fetch per batch during a snapshot read.	No	INTEGER	1024	-
  scan.startup.mode	The startup mode for the connector.	No	STRING	initial	Valid values include: initial (default): On the first startup, the connector performs a snapshot of the existing data and then continues to read the latest Binlog data. latest-offset: On the first startup, the connector skips the snapshot and starts reading from the end of the Binlog. It reads only changes that occur after the connector starts. earliest-offset: The connector skips the snapshot and starts reading from the earliest available Binlog offset. specific-offset: The connector skips the snapshot and starts from a user-specified Binlog offset. You can specify the offset by setting both scan.startup.specific-offset.file and scan.startup.specific-offset.pos, or by setting only scan.startup.specific-offset.gtid-set to start from a specific GTID set. timestamp: The connector skips the snapshot and starts reading the Binlog from a specified timestamp. The timestamp is specified in milliseconds by the scan.startup.timestamp-millis parameter. Important When you use the earliest-offset, specific-offset, or timestamp startup mode, ensure that the table schema does not change between the specified starting offset and the time the job is started. Schema changes can cause errors.
  scan.startup.specific-offset.file	The Binlog filename for the starting offset.	No	STRING	-	When you use this configuration, scan.startup.mode must be set to specific-offset. The file name is in a format such as mysql-bin.000003.
  scan.startup.specific-offset.pos	The starting offset within the specified Binlog file.	No	INTEGER	-	When you use this configuration, scan.startup.mode must be set to specific-offset.
  scan.startup.specific-offset.gtid-set	The GTID set for the starting offset.	No	STRING	-	When you use this configuration, scan.startup.mode must be set to specific-offset. The GTID set is in a format such as 24DA167-0C0C-11E8-8442-00059A3C7B00:1-19.
  scan.startup.timestamp-millis	The starting timestamp in milliseconds.	No	LONG	-	When you use this configuration, scan.startup.mode must be set to timestamp. The unit of the timestamp is milliseconds. Important When starting from a specified timestamp, the MySQL CDC connector scans the initial event of each Binlog file to locate the file corresponding to the timestamp. Ensure that the Binlog file for the given timestamp has not been purged and is accessible on the database server.
  server-time-zone	The session time zone of the database server.	No	STRING	If this parameter is not specified, the system uses the time zone of the Flink job's runtime environment, which is the time zone of the selected availability zone.	Example: Asia/Shanghai. Controls how TIMESTAMP values are converted to strings. Debezium temporal values.
  debezium.min.row.count.to.stream.results	When the number of rows in a table exceeds this value, the connector uses batch read instead of snapshot read.	No	INTEGER	1000	Flink reads MySQL source tables in one of the following ways: Snapshot read: Loads the entire table into memory. This method is fast but memory-intensive and can cause out of memory (OOM) errors for large tables. Batch read: Reads the table in smaller batches. This method is memory-efficient but can be slower than a snapshot read.
  connect.timeout	The timeout for a connection attempt to the MySQL server. If the attempt times out, the connector retries.	No	DURATION	30s	-
  connect.max-retries	The maximum number of times to retry a failed connection to the MySQL server.	No	INTEGER	3	-
  connection.pool.size	The size of the database connection pool.	No	INTEGER	20	The connection pool reuses database connections to reduce the overall connection count.
  jdbc.properties.*	Custom parameters appended to the JDBC URL.	No	STRING	-	You can pass custom connection parameters. For example, to disable SSL, set 'jdbc.properties.useSSL' = 'false'. For a list of supported parameters, see MySQL Configuration Properties.
  debezium.*	Custom Debezium parameters for reading the Binlog.	No	STRING	-	You can pass custom Debezium parameters. For example, use 'debezium.event.deserialization.failure.handling.mode'='ignore' to specify the handling mode for deserialization failures. Warning Do not modify Debezium parameters arbitrarily, as this can cause data reading errors. For example, the debezium.binlog.buffer.size parameter must not be configured.
  heartbeat.interval	The interval for emitting heartbeat events to advance the Binlog offset.	No	DURATION	30s	Heartbeat events advance the Binlog offset. This is especially useful for slowly updated tables where the Binlog offset might not advance automatically. This prevents the Binlog offset from becoming outdated, which could cause the job to fail and require a stateless restart.
  scan.incremental.snapshot.chunk.key-column	Specifies a column to use for splitting chunks during the snapshot phase.	See the Remarks column.	STRING	-	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. You can select only one column from the primary key.
  rds.region-id	The region ID of the ApsaraDB RDS for MySQL instance.	Required when reading archived logs from OSS.	STRING	-	For a list of region IDs, see Regions and zones. Important MySQL CDC generates random GTID strings, which are not monotonically increasing like Binlog file offsets. To locate a specific file by its GTID, the connector must download and parse all archived logs from OSS, a process that is resource-intensive and time-consuming. Therefore, reading from archived OSS logs is not feasible with GTID-based startup modes. This feature only supports starting from a specific timestamp or a Binlog file offset. It also does not support scenarios involving primary-secondary switchovers within the archived logs, as these rely on GTIDs. Evaluate these limitations carefully before using this feature.
  rds.access-key-id	The AccessKey ID for the ApsaraDB RDS for MySQL account.	Required when reading archived logs from OSS.	STRING	-	How do I view my AccessKey ID and AccessKey secret?. Important Security Note: To avoid exposing your credentials, use Variable Management to store and retrieve your AccessKey ID.
  rds.access-key-secret	The AccessKey secret for the ApsaraDB RDS for MySQL account.	Required when reading archived logs from OSS.	STRING	-	How do I view my AccessKey ID and AccessKey secret? Important Security Note: To avoid exposing your credentials, use Variable Management to store and retrieve your AccessKey secret.
  rds.db-instance-id	The instance ID of the ApsaraDB RDS for MySQL instance.	Required when reading archived logs from OSS.	STRING	-	-
  rds.main-db-id	The primary database ID of the ApsaraDB RDS for MySQL instance.	No	STRING	-	To obtain the primary database ID, see ApsaraDB RDS for MySQL log backup. Supported in Flink engine VVR 8.0.7 and later. Note If this parameter is not specified, Flink engine VVR 11.7 and later automatically queries the primary database ID based on the ApsaraDB RDS for MySQL connection information.
  rds.download.timeout	The timeout for downloading a single archived log file from OSS.	No	DURATION	60s	-
  rds.endpoint	The service endpoint for retrieving Binlog information from OSS.	No	STRING	-	For a list of available values, see Service endpoints. Supported in Flink engine VVR 8.0.8 and later.
  scan.incremental.close-idle-reader.enabled	Specifies whether to close an idle reader after the snapshot phase is complete.	No	BOOLEAN	false	Supported in Flink engine VVR 8.0.1 and later. For this setting to take effect, you must also set execution.checkpointing.checkpoints-after-tasks-finish.enabled to true.
  scan.read-changelog-as-append-only.enabled	Specifies whether to convert the changelog stream to an append-only stream.	No	BOOLEAN	false	Valid values: true: All message types (including INSERT, DELETE, UPDATE_BEFORE, and UPDATE_AFTER) are converted to INSERT messages. Enable this only for specific use cases, such as preserving deleted messages from an upstream table. false (default): All message types are emitted as-is. Note Supported in Flink engine VVR 8.0.8 and later.
  scan.only.deserialize.captured.tables.changelog.enabled	During the incremental phase, specifies whether to deserialize change events only for the captured tables.	No	BOOLEAN	false for Flink engine VVR 8.x versions. true for Flink engine VVR 11.1 and later.	Valid values: true: Deserializes change data only for the captured tables, which improves Binlog reading speed. false: Deserializes change data for all tables. Note Supported in Flink engine VVR 8.0.7 and later. In Flink engine VVR 8.0.8 and earlier versions, this parameter is named debezium.scan.only.deserialize.captured.tables.changelog.enable.
  scan.parse.online.schema.changes.enabled	During the incremental phase, specifies whether to parse DDL events from RDS lockless change.	No	BOOLEAN	false	Valid values: true: Parses DDL events from RDS lockless change. false (default): Does not parse DDL events from RDS lockless change. This is an experimental feature. Take a savepoint of your Flink job before you perform online lockless changes to ensure you can recover if needed. Note Supported in Flink engine VVR 11.1 and later.
  scan.incremental.snapshot.backfill.skip	Specifies whether to skip the backfill process during the snapshot phase.	No	BOOLEAN	false	Valid values: true: Skips the backfill process. false (default): Does not skip the backfill process. If backfill is skipped, changes to the table that occur during the snapshot phase are read later in the incremental phase, instead of being merged into the snapshot. Important Skipping backfill can lead to data inconsistencies because changes that occur during the snapshot phase might be replayed. This provides only at-least-once semantics. Note Supported in Flink engine VVR 11.1 and later.
  scan.incremental.snapshot.unbounded-chunk-first.enabled	Specifies whether to distribute the unbounded chunk first during the snapshot phase.	No	BOOLEAN	false	Valid values: true: The unbounded chunk is prioritized for distribution during the snapshot phase. false (default): The unbounded chunk is not prioritized. This is an experimental feature. Enabling it can reduce the risk of out of memory (OOM) errors on the TaskManager when it synchronizes the final chunk. Enable this feature before the job starts for the first time. Note Supported in Flink engine VVR 11.1 and later.
  binlog.session.network.timeout	The network read and write timeout for the Binlog connection.	No	DURATION	10m	If set to 0s, the connection uses the default timeout of the MySQL server. Note Supported in Flink engine VVR 11.5 and later.
  scan.rate-limit.records-per-second	Limits the maximum number of records that the source can emit per second.	No	LONG	-	Use this parameter to limit the data read rate. This limit applies to both the snapshot and incremental phases. The numRecordsOutPerSecond metric of the Source reflects the number of records output per second for the entire data stream. You can adjust this parameter based on this metric. During the snapshot read phase, you typically need to reduce the number of data rows read in each batch by reducing the value of the scan.incremental.snapshot.chunk.size parameter. Note Supported in Flink engine VVR 11.5 and later.
  scan.binlog.tolerate.gtid-holes	Enables the job to ignore gaps in the GTID sequence, allowing it to bypass discontinuous events and continue running.	No	BOOLEAN	false	Before enabling this parameter, ensure that the startup offset of the job has not expired. If the job starts from a purged or expired GTID offset, the engine silently skips the missing logs, causing data loss. Note Supported in Flink engine VVR 11.6 and later.

• Dimension table parameters

  Parameter	Description	Required	Type	Default	Notes
  url	The JDBC URL for the MySQL database.	No	STRING	None	The URL format is: jdbc:mysql://<endpoint>:<port>/<database name>.
  lookup.max-retries	The maximum number of retries if a lookup request fails.	No	INTEGER	3	Supported only by Flink compute engine VVR 6.0.7 and later.
  lookup.cache.strategy	The caching policy.	No	STRING	None	Supported values: None, LRU, and ALL. Background information. Note The LRU caching policy requires the lookup.cache.max-rows parameter.
  lookup.cache.max-rows	The maximum number of rows to store in the cache.	No	INTEGER	100000	Required if the caching policy is set to LRU. Optional if the caching policy is set to ALL.
  lookup.cache.ttl	The time-to-live (TTL) for cached entries.	No	DURATION	None	The behavior of the lookup.cache.ttl parameter depends on the value of lookup.cache.strategy: If lookup.cache.strategy is set to None, lookup.cache.ttl does not need to be configured, which means the cache does not expire. If lookup.cache.strategy is set to LRU, lookup.cache.ttl specifies the cache timeout duration. By default, the cache does not expire. If lookup.cache.strategy is set to ALL, lookup.cache.ttl specifies the cache reload interval. By default, the cache is not reloaded. Specify the duration in a time format, such as 1min or 10s.
  lookup.max-join-rows	The maximum number of matching rows to return from the dimension table for each row in the source table.	No	INTEGER	1024	None
  lookup.filter-push-down.enabled	Specifies whether to enable filter pushdown for the dimension table.	No	BOOLEAN	false	Valid values: true: Enables filter pushdown. The system filters data from the dimension table based on the conditions in the SQL job before loading it. false (Default): Disables filter pushdown. The system loads the full dataset from the dimension table. Note Supported only by Flink compute engine VVR 8.0.7 and later. Important Use filter pushdown only for Flink dimension tables. This feature is not supported for MySQL source tables. If you use the same Flink table as both a source and a dimension table with filter pushdown enabled, you must use SQL Hints to set this parameter to false when using it as a source table. Otherwise, the job may fail.

• Specific to sink tables

  Parameter	Description	Required	Type	Default	Notes
  url	The JDBC URL for the MySQL database.	No	STRING	None	The format of the URL is: jdbc:mysql://<endpoint>:<port>/<database name>.
  sink.max-retries	The maximum number of retries for a failed write operation.	No	INTEGER	3	None
  sink.buffer-flush.batch-size	The number of records to include in each batch write.	No	INTEGER	4096	None
  sink.buffer-flush.max-rows	The maximum number of records to buffer in memory before flushing.	No	INTEGER	10000	This parameter takes effect only when a primary key is specified.
  sink.buffer-flush.interval	The flush interval. The system automatically flushes the buffer when this interval is reached, even if other flush conditions are not met.	No	DURATION	1s	None
  sink.ignore-delete	Whether to ignore DELETE operations.	No	BOOLEAN	false	When a Flink SQL job generates a stream that includes retractions (DELETE or UPDATE_BEFORE messages), concurrent updates to the same table by multiple sink tasks can cause data inconsistencies. For example, if one task deletes a record while another task updates only a few of its fields, the non-updated fields might be incorrectly set to null or their default values. To prevent this issue, set sink.ignore-delete to true to ignore upstream DELETE and UPDATE_BEFORE operations. Note UPDATE_BEFORE is part of Flink's retraction mechanism, which retracts the old value during an update operation. When sink.ignore-delete is true, the sink skips all DELETE and UPDATE_BEFORE records, processing only INSERT and UPDATE_AFTER records.
  sink.ignore-null-when-update	How to handle null values in update records.	No	BOOLEAN	false	Valid values: true: Ignores null values, leaving the corresponding field in the destination table unchanged. This setting requires a primary key to be defined for the Flink table. When set to true: For VVR 8.0.6 and earlier, batch write is not supported for the sink table. For VVR 8.0.7 and later, batch write is supported for the sink table. While a batch write can significantly improve throughput, it may introduce data latency and increase the risk of out-of-memory errors. Evaluate these trade-offs based on your business requirements. false: Updates the corresponding field to null. Note This parameter is supported in VVR versions 8.0.5 and later.

Type mapping

• CDC source table

  MySQL CDC type	Flink type
  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

  Important

  Avoid using the MySQL TINYINT(1) type to store values other than 0 and 1. By default, when property-version is 0, the MySQL CDC source table maps TINYINT(1) to the Flink BOOLEAN type, which can lead to data inaccuracies. To use the TINYINT(1) type to store other numeric values, see the catalog.table.treat-tinyint1-as-boolean configuration parameter.

• Dimension and result tables

  MySQL type	Flink type
  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) Note p must be 38 or less.
  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 Important Flink supports MySQL BLOB records up to 2,147,483,647 bytes (2^31 - 1).
  BLOB
  MEDIUMBLOB
  LONGBLOB

Data ingestion

You can use the MySQL connector as a data source for Flink CDC.

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

Parameters

Reuse an existing catalog

Starting with VVR 11.5, you can directly reference a built-in MySQL catalog from the Catalogs page in a Flink CDC data ingestion job, eliminating the need to manually configure connection properties.

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

Currently, data ingestion jobs automatically reuse the following parameters from the MySQL catalog:

• hostname

• port

• username

• password

• catalog.table.metadata-columns

• catalog.table.treat-tinyint1-as-boolean

To override any of these parameters, explicitly define them in your YAML configuration. Parameters defined in YAML take precedence.

Type mapping

The following table shows the type mappings for data ingestion.