Realtime Compute for Apache Flink:MySQL connector best practices

Set the server ID to avoid binary log conflicts

Each MySQL CDC client requires a unique server ID. Multiple source tables sharing the same server ID cause conflicts and job errors. Set a server ID in the Flink table's DDL statement or via SQL Hints. Using SQL Hints is recommended.

Configure the server ID based on your deployment scenario:

• Incremental snapshot disabled, or parallelism = 1: Specify a single server ID.

  SELECT * FROM source_table /*+ OPTIONS('server-id'='123456') */ ;

• Incremental snapshot enabled and parallelism > 1: Specify a range. The range must contain at least as many IDs as the parallelism value. For example, with parallelism = 3:

  SELECT * FROM source_table /*+ OPTIONS('server-id'='123456-123458') */ ;

• CTAS-based synchronization: MySQL CDC source tables with identical configurations are automatically merged. Assign the same server ID to all merged source tables. For details, see the "Example 4: execution of multiple CREATE TABLE AS statements" section of the CREATE TABLE AS statement topic.

• Multiple MySQL CDC source tables without CTAS: If source reuse is disabled (see Enable source reuse to reduce binary log connections), assign a unique server ID to each source table. If incremental snapshot is enabled and parallelism > 1, also specify a range per table:

  SELECT * FROM
    source_table1 /*+ OPTIONS('server-id'='123456-123457') */
  LEFT JOIN
    source_table2 /*+ OPTIONS('server-id'='123458-123459') */
  ON source_table1.id = source_table2.id;

Configure chunk options for memory optimization

When the MySQL CDC source connector starts, it scans the entire table, splits data into chunks by primary key, and records the binary log position. Checkpoints track progress at chunk granularity. On failover, the connector resumes from the last unread chunk. After all chunks are read, the connector switches to incremental reading from the recorded binary log position, ensuring exactly-once semantics.

For a detailed description of the incremental snapshot algorithm, see MySQL CDC connector in the Apache Flink documentation.

By default, tables with a single-column primary key split by that key; tables with a composite primary key split by the first column. Ververica Runtime (VVR) 6.0.7 and later supports tables without a primary key — use scan.incremental.snapshot.chunk.key-column to specify a non-null column as the chunk key.

Accelerate snapshot reads

During the snapshot phase, the MySQL source connector reads data through a JDBC connection. Use the following methods to speed up snapshot reads:

1. Increase the source parallelism.

2. Increase scan.incremental.snapshot.chunk.size to fetch more rows per chunk.

3. If the downstream sink supports idempotent writes with a primary key, enable scan.incremental.snapshot.backfill.skip to skip binary log reads during backfill.

Enable source reuse to reduce binary log connections

Source reuse merges multiple MySQL CDC source tables in a job into a single binary log connection, reducing database load. This feature is available only in the Realtime Compute for Apache Flink connector for MySQL CDC.

Enable source reuse in an SQL job:

SET 'table.optimizer.source-merge.enabled' = 'true';

Enable this on new jobs only. On an existing job, enabling source reuse changes the job topology — a stateful restart may fail or cause data loss. Perform a stateless restart instead.

After enabling, MySQL CDC source tables with identical configurations are merged. When all source tables share the same configuration, MySQL connection counts are:

Accelerate incremental reading

The MySQL CDC connector parses binary log files to generate change messages. Because binary logs record changes across all tables, parsing can become a bottleneck when many tables are involved.

CREATE TABLE mysql_source (...) WITH (
    'connector' = 'mysql-cdc',
    -- Debezium options
    'debezium.max.queue.size' = '162580',
    'debezium.max.batch.size' = '40960',
    'debezium.poll.interval.ms' = '50',
    -- Parallel parsing and event filtering
    'scan.only.deserialize.captured.tables.changelog.enabled' = 'true',  -- Parse only the specified tables
    'scan.parallel-deserialize-changelog.enabled' = 'true'               -- Use multiple threads for parsing
    ...
)

Diagnose data latency

If data latency occurs during the incremental phase, follow these steps to identify and resolve the bottleneck.

Read RDS binary logs from OSS to prevent expiration

When using ApsaraDB RDS for MySQL as a data source, Flink can read binary log backups stored in OSS (Object Storage Service). If the requested binary log file is in OSS, Flink downloads it to the cluster before processing. If the file is still on the database server, Flink reads it directly through the database connection. This feature is available only in the Realtime Compute for Apache Flink connector for MySQL CDC.

To enable this, add the following options to your table definition:

CREATE TABLE mysql_source (...) WITH (
    'connector' = 'mysql-cdc',
    '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',   -- ApsaraDB RDS for MySQL instance ID
    'rds.main-db-id' = '12345678',           -- Primary database ID
    'rds.endpoint' = 'rds.aliyuncs.com'
    ...
)

Synchronize database data and schema changes with data ingestion

For data synchronization tasks, create a data ingestion deployment, which is optimized for data integration scenarios. See Getting started with a YAML deployment for data ingestion and Develop a Flink CDC job (Beta).

The following example synchronizes data and schema changes from the MySQL database app_db to Hologres:

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

Control new table discovery

The MySQL data ingestion connector provides two options for controlling how new tables are discovered during a job run: