Realtime Compute for Apache Flink:MySQL

SQL

You can use the MySQL connector in SQL jobs as a source table, dimension table, 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	Data type	Default value	Remarks
  connector	The type of the table.	Yes	STRING	None	When used as a source table, set this option to mysql-cdc or mysql. They are equivalent. When used as a dimension or sink table, set this option to mysql.
  hostname	The IP address or hostname of the MySQL database.	Yes	STRING	None	We recommend entering the 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 the public network for access. For more information, see Manage and operate workspaces and How can a fully managed Flink cluster access the public network?.
  username	The username for the MySQL database service.	Yes	STRING	None	None.
  password	The password for the MySQL database service.	Yes	STRING	None	None.
  database-name	The name of the MySQL database.	Yes	STRING	None	When you use a database as a source table, you can use a regular expression for the database name to read data from multiple databases. When you use a regular expression, avoid using the ^ and $ symbols to match the start and end of the string. For more information, see the Remarks column for the table-name option.
  table-name	The name of the MySQL table.	Yes	STRING	None	You can use a regular expression for the source table name to read data from multiple tables. When you read data from multiple MySQL tables, submit multiple CTAS statements as a single job. This avoids enabling multiple binary log listeners and improves performance and efficiency. For more information, see Multiple CTAS statements: Submit as a single job. When you use a regular expression, avoid using the ^ and $ symbols to match the start and end of the string. For more information, see the following note. Note When a MySQL CDC source table matches table names, it combines the database-name and table-name that you specify into a full-path regular expression using the string \\. (the character . is used in VVR versions earlier than 8.0.1). It then uses this regular expression to match the fully qualified names of tables in the MySQL database. For example, if you set 'database-name'='db_.*' and 'table-name'='tb_.+', the connector uses the regular expression db_.*\\.tb_.+ (db_.*.tb_.+ in versions earlier than 8.0.1) to match the fully qualified table names to determine which tables to read.
  port	The port number of the MySQL database service.	No	INTEGER	3306	None.

• Source-specific

  Parameter	Description	Required	Data type	Default value	Remarks
  server-id	A numeric ID for the database client.	No	STRING	A random value between 5400 and 6400 is generated.	This ID must be globally unique within the MySQL cluster. We recommend that you set a different ID for each job that connects to the same database. This option also supports an ID range, such as 5400-5408. 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. For more information, see Use of Server ID.
  scan.incremental.snapshot.enabled	Specifies whether to enable incremental snapshots.	No	BOOLEAN	true	Incremental snapshot is enabled by default. Incremental snapshot is a new mechanism for reading full data snapshots. Compared to the old snapshot reading method, incremental snapshots offer several advantages: The source can read full data in parallel. The source supports chunk-level checkpoints when reading full data. The source does not need to acquire a global read lock (FLUSH TABLES WITH read lock) when reading full data. If you want the source to support concurrent reading, each concurrent reader needs a unique server ID. Therefore, server-id must be a range, such as 5400-6400, and the range must be greater than or equal to the degree of parallelism. Note This configuration item is removed in Ververica Runtime (VVR) 11.1 and later.
  scan.incremental.snapshot.chunk.size	The size of each chunk in number of rows.	No	INTEGER	8096	When incremental snapshot reading is enabled, the table is split into multiple chunks for reading. The data of a chunk is buffered in memory before it is fully read. The fewer rows a chunk contains, the greater the total number of chunks in the table. Although this reduces the granularity of fault recovery, it can cause out-of-memory (OOM) issues and reduce overall throughput. Therefore, you need to weigh the trade-offs and set a reasonable chunk size.
  scan.snapshot.fetch.size	The maximum number of records to fetch 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): Scans the full historical data first and then reads the latest binlog data upon the first startup. latest-offset: Does not scan historical data upon the first startup. It starts reading from the end of the binlog, which means it only reads the latest changes made after the connector starts. earliest-offset: Does not scan historical data. It starts reading from the earliest available binlog. specific-offset: Does not scan historical data. It starts from a specific binlog offset that you specify. 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: Does not scan historical data. It starts reading the binlog from a specified timestamp. The timestamp is specified by scan.startup.timestamp-millis in milliseconds. Important When you use the earliest-offset, specific-offset, or timestamp startup mode, make sure that the schema of the corresponding table does not change between the specified binlog consumption position and the job startup time to avoid errors due to schema mismatch.
  scan.startup.specific-offset.file	The binlog filename of the start offset when using the specific offset startup mode.	No	STRING	None	When you use this configuration, you must set scan.startup.mode to specific-offset. The filename format is, for example, mysql-bin.000003.
  scan.startup.specific-offset.pos	The offset in the specified binlog file to start from when using the specific offset startup mode.	No	INTEGER	None	When you use this configuration, you must set scan.startup.mode to specific-offset.
  scan.startup.specific-offset.gtid-set	The GTID set of the start offset when using the specific offset startup mode.	No	STRING	None	When you use this configuration, you must set scan.startup.mode to specific-offset. The GTID set format is, for example, 24DA167-0C0C-11E8-8442-00059A3C7B00:1-19.
  scan.startup.timestamp-millis	The start offset as a millisecond timestamp when using the timestamp startup mode.	No	LONG	None	When you use this configuration, you must set scan.startup.mode to timestamp. The timestamp is in milliseconds. Important When you specify a time, MySQL CDC attempts to read the initial event of each binlog file to determine its timestamp and locate the binlog file corresponding to the specified time. Make sure that the binlog file for 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 option, the system uses the time zone of the Flink job's runtime environment as the database server time zone. This is the time zone of the zone you selected.	Example: Asia/Shanghai. This option controls how TIMESTAMP types in MySQL are converted to STRING types. For more information, see Debezium temporal values.
  debezium.min.row.count.to.stream.results	When the number of rows in a table exceeds this value, batch reading mode is used.	No	INTEGER	1000	Flink reads data from a MySQL source table in the following ways: Full read: Reads the entire table's data directly into memory. This method is fast but consumes a corresponding amount of memory. If the source table is very large, it may pose an OOM risk. Batch read: Reads data in multiple batches, with a certain number of rows per batch, until all data is read. This method avoids OOM risks when reading large tables but is relatively slower.
  connect.timeout	The maximum time to wait for a connection to the MySQL database server to time out before retrying.	No	DURATION	30s	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.
  jdbc.properties.*	Custom connection options in the JDBC URL.	No	STRING	None	You can pass custom connection options. For example, to not use the SSL protocol, you can set 'jdbc.properties.useSSL' = 'false'. For more information about supported connection options, see MySQL Configuration Properties.
  debezium.*	Custom options for Debezium to read binlogs.	No	STRING	None	You can pass custom Debezium options. For example, use 'debezium.event.deserialization.failure.handling.mode'='ignore' to specify the handling logic for parsing errors.
  heartbeat.interval	The interval at which the source uses heartbeat events to advance the binlog offset.	No	DURATION	30s	Heartbeat events are used to advance the binlog offset in the source, which is very useful for slowly updated tables in MySQL. For such tables, the binlog offset does not advance automatically. Heartbeat events can push the binlog offset forward, preventing issues where an expired binlog offset causes the job to fail and require a stateless restart.
  scan.incremental.snapshot.chunk.key-column	The column used to split chunks during the snapshot phase.	See Remarks.	STRING	None	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 only select 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	None	For more information about region IDs, see Regions and zones.
  rds.access-key-id	The AccessKey ID of the account for the ApsaraDB RDS for MySQL instance.	Required when reading 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, we recommend that you manage your AccessKey ID using secrets management. For more information, see Manage variables.
  rds.access-key-secret	The AccessKey secret of the account for the ApsaraDB RDS for MySQL instance.	Required when reading 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, we recommend that you manage your AccessKey secret using secrets management. For more information, see Manage variables.
  rds.db-instance-id	The instance ID of the ApsaraDB RDS for MySQL instance.	Required when reading archived logs from OSS.	STRING	None	None.
  rds.main-db-id	The ID of the primary database of the ApsaraDB RDS for MySQL instance.	No	STRING	None	For more information about how to obtain the primary database ID, see Log backup for ApsaraDB RDS for MySQL. Supported only in VVR 8.0.7 and later.
  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 binlog information.	No	STRING	None	For more information about valid values, see Endpoints. Supported only in VVR 8.0.8 and later.
  scan.incremental.close-idle-reader.enabled	Specifies whether to close idle readers after the snapshot phase ends.	No	BOOLEAN	false	Supported only in VVR 8.0.1 and later. For this configuration to take effect, you must 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 types of messages, including INSERT, DELETE, UPDATE_BEFORE, and UPDATE_AFTER, are converted to INSERT messages. Enable this only in special scenarios, such as when you need to save delete messages from the upstream table. false (Default): All types of messages are sent downstream as they are. Note Supported only in 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 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 change data only for the target tables to accelerate binlog reading. false (Default): Deserializes change data for all tables. Note Supported only in VVR 8.0.7 and later. When used in VVR 8.0.8 and earlier, the option name must be changed to debezium.scan.only.deserialize.captured.tables.changelog.enable.
  scan.parse.online.schema.changes.enabled	During 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. We recommend that you take a snapshot of the Flink job for recovery before you perform online lockless changes. Note Supported only in VVR 11.1 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. If you skip backfill, changes to the table during the snapshot phase are read in the later incremental phase instead of being merged into the snapshot. Important Skipping backfill may lead to data inconsistency because changes that occur during the snapshot phase might be replayed. Only at-least-once semantics are guaranteed. Note Supported only in VVR 11.1 and later.
  scan.incremental.snapshot.unbounded-chunk-first.enabled	Specifies whether to distribute unbounded chunks first during the snapshot reading phase.	No	BOOLEAN	false	Valid values: true: Distributes unbounded chunks first during the snapshot reading phase. false (Default): Does not distribute unbounded chunks first during the snapshot reading phase. This is an experimental feature. Enabling it can reduce the risk of out-of-memory (OOM) errors when a TaskManager synchronizes the last chunk during the snapshot phase. We recommend that you add this before the job's first startup. Note Supported only in VVR 11.1 and later.
  binlog.session.network.timeout	The network read/write timeout for the binlog connection.	No	DURATION	10m	If you set this option 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	The maximum number of records that the source can send per second.	No	LONG	None	This option applies to scenarios where you need to limit data reading. The limit takes effect in both the full and incremental phases. The numRecordsOutPerSecond metric of the source reflects the number of records that the data stream outputs per second. You can adjust this option based on this metric. In the full data reading phase, you typically need to reduce the number of data records read in each batch. You can reduce the value of the scan.incremental.snapshot.chunk.size option. Note Supported only in VVR 11.5 and later.

• Dimension table-specific

  Parameter	Description	Required	Data type	Default value	Remarks
  url	The MySQL JDBC URL.	No	STRING	None	The URL format is: jdbc:mysql://<endpoint>:<port>/<database_name>.
  lookup.max-retries	The maximum number of retries after a failed data read.	No	INTEGER	3	Supported only in VVR 6.0.7 and later.
  lookup.cache.strategy	The cache policy.	No	STRING	None	Supports three cache policies: None, LRU, and ALL. For more information about the values, see Background information. Note When you use the LRU cache policy, you must also configure the lookup.cache.max-rows option.
  lookup.cache.max-rows	The maximum number of cached rows.	No	INTEGER	100000	If you select the LRU cache policy, you must set the cache size. If you select the ALL cache policy, you do not need to set the cache size.
  lookup.cache.ttl	The cache time-to-live (TTL).	No	DURATION	10 s	The configuration of lookup.cache.ttl depends on lookup.cache.strategy: If lookup.cache.strategy is set to None, you do not need to configure lookup.cache.ttl. This means the cache does not expire. If lookup.cache.strategy is set to LRU, lookup.cache.ttl is the cache TTL. By default, it does not expire. If lookup.cache.strategy is set to ALL, lookup.cache.ttl is the cache reload time. By default, it is not reloaded. Use a time format, such as 1min or 10s.
  lookup.max-join-rows	The maximum number of results returned when querying the dimension table for each record in the primary 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 for the dimension table. When loading data from the MySQL database table, the dimension table filters data in advance based on the conditions set in the SQL job. false (Default): Disables filter pushdown for the dimension table. When loading data from the MySQL database table, the dimension table loads all data. Note Supported only in VVR 8.0.7 and later. Important Filter pushdown for a dimension table should only be enabled when the Flink table is used as a dimension table. MySQL source tables do not support enabling filter pushdown. If a Flink table is used as both a source table and a dimension table, and filter pushdown is enabled for the dimension table, you must explicitly set this configuration item to false for the source table using SQL Hints. Otherwise, the job may run abnormally.

• Sink-specific

  Parameter	Description	Required	Data type	Default value	Remarks
  url	The MySQL JDBC URL.	No	STRING	None	The URL format is: jdbc:mysql://<endpoint>:<port>/<database_name>.
  sink.max-retries	The maximum number of retries after a failed data write.	No	INTEGER	3	None.
  sink.buffer-flush.batch-size	The number of records in a single batch write.	No	INTEGER	4096	None.
  sink.buffer-flush.max-rows	The number of data records buffered in memory.	No	INTEGER	10000	This option takes effect only after a primary key is specified.
  sink.buffer-flush.interval	The time interval for flushing the buffer. If the data in the buffer does not meet the output conditions after the specified waiting time, the system automatically outputs all data in the buffer.	No	DURATION	1s	None.
  sink.ignore-delete	Specifies whether to ignore DELETE operations.	No	BOOLEAN	false	When the stream generated by Flink SQL contains delete or update-before records, if multiple output tasks update different fields of the same table simultaneously, data inconsistency may occur. For example, after a record is deleted, another task updates only some fields. The un-updated fields will become null or their default values, causing data errors. By setting sink.ignore-delete to true, you can ignore upstream DELETE and UPDATE_BEFORE operations to avoid such issues. Note UPDATE_BEFORE is part of Flink's retraction mechanism, used to "retract" the old value in an update operation. When ignoreDelete = true, all DELETE and UPDATE_BEFORE records are skipped. Only INSERT and UPDATE_AFTER records are processed.
  sink.ignore-null-when-update	When updating data, specifies whether to update the corresponding field to null or skip the update for that field if the incoming data field is null.	No	BOOLEAN	false	Valid values: true: Does not update the field. This option can be set to true only when a primary key is set for the Flink table. When set to true: In VVR 8.0.6 and earlier, batch execution is not supported for writing data to the sink table. In VVR 8.0.7 and later, batch execution is supported for writing data to the sink table. Although batch writing can significantly improve write efficiency and overall throughput, it can cause data latency and OOM risks. Therefore, make a trade-off based on your actual business scenario. false: Updates the field to null. Note This option is supported only in VVR 8.0.5 and later.

Type mapping

• CDC source table

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

  We recommend that you do not use the TINYINT(1) type in MySQL to store values other than 0 and 1. When property-version = 0, the MySQL CDC source table maps TINYINT(1) to Flink's BOOLEAN type by default, which can cause data inaccuracies. To use the TINYINT(1) type to store values other than 0 and 1, see the configuration option catalog.table.treat-tinyint1-as-boolean.

• Dimension table and sink table

  MySQL field type	Flink field 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 less than or equal to 38.
  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 type records only up to 2,147,483,647 (2^31 - 1) bytes.
  BLOB
  MEDIUMBLOB
  LONGBLOB

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

Type mapping

The following table shows the type mappings for data ingestion.