MySQL SQL connector

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	Notes
  connector	The table type.	Yes	STRING	None	When used as a source table, you can set this parameter to mysql-cdc or mysql. They are equivalent. When used as a dimension table or sink table, the value must be mysql.
  hostname	The IP address or hostname of the MySQL database.	Yes	STRING	None	We recommend that you specify a virtual private cloud (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.
  database-name	The name of the MySQL database.	Yes	STRING	None	When a database is used as a source table, you can use a regular expression for the database name to read data from multiple databases. When you use regular expressions, do not use the ^ and $ symbols to match the beginning and end of the string. For more information, see the notes for the table-name parameter.
  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 regular expressions, do not use the ^ and $ symbols to match the beginning and end of the string. For more information, see the following note. Note When a MySQL CDC source table matches table names using a regular expression, it concatenates the database-name and table-name that you specify with the string \\. to form a full-path regular expression. Before VVR 8.0.1, the character . was used. The connector 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_.+ to match the fully qualified table names to determine which tables to read. Before VVR 8.0.1, the regular expression was db_.*.tb_.+.
  port	The port number of the MySQL database service.	No	INTEGER	3306	None.

• Source table only

  Parameter	Description	Required	Data type	Default value	Notes
  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. 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. When incremental reading is enabled, concurrent reading is supported. In this case, set an ID range so that each concurrent reader uses a different ID. For more information, see Use Server ID.
  scan.incremental.snapshot.enabled	Specifies whether to enable incremental snapshots.	No	BOOLEAN	true	Incremental snapshots are enabled by default. Incremental snapshot is a new mechanism for reading full data snapshots. Compared to the old snapshot reading method, incremental snapshots have many advantages, including the following: 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 size of the range must be greater than or equal to the concurrency. 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 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 out-of-memory (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 When using the earliest-offset, specific-offset, or timestamp startup mode, ensure that the schema of the corresponding table does not change between the specified binary log consumption position and the job startup time. This prevents errors caused by schema mismatches.
  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.
  debezium.min.row.count.to.stream.results	When the number of rows in a table is greater than this value, batch reading mode is used.	No	INTEGER	1000	Flink reads data from a MySQL source table in one of 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, there is a risk of OOM errors. 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 slow.
  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 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.
  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.
  scan.incremental.snapshot.chunk.key-column	Specifies a column to be used as the splitting column for sharding during the snapshot phase.	See the Notes column.	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. Only one column can be selected from the primary key.
  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. Supported only in VVR 8.0.7 and later. 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. Supported only in VVR 8.0.8 and later.
  scan.incremental.close-idle-reader.enabled	Specifies whether to close idle readers after the snapshot is complete.	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 data stream to an append-only data 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 option 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	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. Note Supported only in VVR 8.0.7 and later. When you use VVR 8.0.8 or earlier, you must change the parameter name to debezium.scan.only.deserialize.captured.tables.changelog.enable.
  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.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. 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.
  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 read/write 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.
  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.

• Dimension table-specific parameters

  Parameter	Description	Required	Data type	Default value	Notes
  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	The supported cache policies are None, LRU, and ALL. For more information about the values, see Dimension table JOIN statements. Note When you use the LRU cache policy, you must also configure the lookup.cache.max-rows parameter.
  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 time out. If lookup.cache.strategy is set to LRU, lookup.cache.ttl is the cache TTL. By default, the cache does not expire. If lookup.cache.strategy is set to ALL, lookup.cache.ttl is the cache loading time. By default, the cache is not reloaded. Use a time format, such as 1min or 10s.
  lookup.max-join-rows	The maximum number of results returned when a record from the primary table matches records in the dimension 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 Dimension table pushdown should only be enabled when a 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.

• For sink tables only

  Parameter	Description	Required	Data type	Default value	Notes
  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 rows in a single batch write.	No	INTEGER	4096	None.
  sink.buffer-flush.max-rows	The number of data rows cached in memory.	No	INTEGER	10000	This parameter takes effect only after a primary key is specified.
  sink.buffer-flush.interval	The interval for flushing the cache. If the data in the cache does not meet the output conditions after the specified waiting time, the system automatically outputs all data in the cache.	No	DURATION	1s	None.
  sink.ignore-delete	Specifies whether to ignore data DELETE operations.	No	BOOLEAN	false	When the stream generated by Flink SQL includes 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 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 type records are skipped. Only INSERT and UPDATE_AFTER records are processed.
  sink.ignore-delete-mode	The strategy for handling delete-type records after DELETE operations are ignored.	No	STRING	ALL	Valid values: ALL: Ignore both -D and -U records. REAL_DELETE: Ignore only -D records. UPDATE_BEFORE: Ignore only -U records. Note This option is supported only in Realtime Compute engine VVR 11.8 and later. Effective only when sink.ignore-delete=true. Configuring it alone results in an error.
  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 value is null.	No	BOOLEAN	false	Valid values: true: Does not update the field. This parameter can be set to true only when a primary key is set for the Flink table. When set to true: For VVR 8.0.6 and earlier, the sink table does not support batch writing. For VVR 8.0.7 and later, the sink table supports batch writing. Batch writing can significantly improve write efficiency and overall throughput, but it introduces data latency and the risk of OOM errors. Therefore, you must make a trade-off based on your business scenario. false: Updates the field to null. Note This parameter is supported only in VVR 8.0.5 and later.

Type mapping

• CDC source tables

  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

  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 the BOOLEAN type in Flink by default. This can cause data inaccuracies. To use the TINYINT(1) type to store values other than 0 and 1, see the configuration parameter catalog.table.treat-tinyint1-as-boolean.

• Dimension tables and sink tables

  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 where p <= 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 only supports MySQL BLOB type records that are less than or equal to 2,147,483,647 (2^31 - 1) bytes.
  BLOB
  MEDIUMBLOB
  LONGBLOB

Diagnose data latency to optimize job throughput

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

1. 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.

2. 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.

3. 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. You can also enable mini-batch parameters to improve throughput. For more information, see High-Performance Flink SQL Optimization Techniques.

4. If a SinkUpsertMaterializer operator with a large state exists in the job, it can affect read performance. Consider increasing the job parallelism or avoiding the SinkUpsertMaterializer operator. For more information, see Avoid using SinkUpsertMaterializer. Removing the SinkUpsertMaterializer operator from an existing job requires a stateless restart. This is because the job topology changes, and starting from the existing state may cause the job to fail or result in data loss.

Chunk parameter optimization

Chunk data and metadata are stored in memory, which can sometimes lead to out-of-memory (OOM) errors. You can adjust parameters based on which component experiences the OOM error:

• JobManager

  The JobManager stores metadata for all chunks. An excessive number of chunks can cause an OOM error. To resolve this, increase the value of scan.incremental.snapshot.chunk.size to reduce the number of chunks. Alternatively, you can increase the JobManager heap memory by setting jobmanager.memory.heap.size in your runtime configuration. For more information, see Flink Parameter Configuration.

• TaskManager

  • The TaskManager reads the data for each chunk. If a chunk contains too many rows, an OOM error can occur. To resolve this, decrease the value of scan.incremental.snapshot.chunk.size to reduce the number of rows per chunk. You can also increase the TaskManager heap memory by increasing the TaskManager Memory value in your runtime configuration.

  • In VVR 8.0.8 and earlier, the last chunk may contain a large amount of data, which can cause the TaskManager to experience an OOM error. We recommend upgrading to VVR 8.0.9 or later to avoid this issue.

  • For a MySQL CDC source table with a composite primary key, chunks are split based on the first column of the key by default. If data is significantly skewed, with many rows sharing the same value in that column, the chunk for that value can become very large and cause an OOM error in the TaskManager. You can set scan.incremental.snapshot.chunk.key-column to specify a different column from the primary key for splitting.