MySQL connector best practices

Set server ID to avoid binlog conflicts

Each client that synchronizes data from a database has a unique ID called the server ID. If different jobs use the same server ID, conflicts can occur and cause job failures. We recommend that you assign a different server ID for each MySQL CDC data source.

• How to configure the server ID

  You can specify the server ID in a Flink table DDL statement or by using SQL Hints.

  We recommend using SQL Hints to configure the server ID instead of specifying it in the WITH clause of the table DDL. For more information, see SQL Hints.

• server ID configuration for different scenarios

  • Incremental snapshot disabled or parallelism is 1

    If incremental snapshot is disabled or the parallelism is 1, you can specify a single server ID.

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

  • Incremental snapshot enabled and parallelism is greater than 1

    When the incremental snapshot is enabled and the parallelism is greater than 1, you must specify a server ID range. The number of available server IDs in the range must be at least the parallelism. For example, if the parallelism is 3, you can use the following configuration:

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

  • Data synchronization with CTAS

    When you synchronize data by using CREATE TABLE AS (CTAS), CDC data sources with identical configurations are automatically merged and reused. In this case, you can assign the same server ID to multiple CDC data sources. For more information, see Example 4: Multiple CTAS Statements.

  • A job with multiple MySQL CDC source tables (non-CTAS)

    If a job contains multiple MySQL CDC source tables, does not use CTAS statements, and has source reuse disabled, you must provide a different server ID for each CDC source table. For more information, see Enable source reuse to reduce binary log connections. Similarly, if the incremental snapshot is enabled and the parallelism is greater than 1, you must specify a server ID range.

    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;

Set chunk parameters to optimize memory usage

When a MySQL CDC source table starts, it performs a full table scan, splits the table into multiple chunks based on the primary key, and records the current binary log position. The job then uses an incremental snapshot algorithm to read data from each chunk sequentially by using SELECT statements. The job periodically performs a checkpoint to record completed chunks. In the event of a failover, it resumes reading from the first unfinished chunk. After all chunks are read, the job switches to reading incremental changes from the previously recorded binary log position. The Flink job performs periodic checkpoints to save the binary log position. If a failover occurs, the job resumes processing from the last saved position, thus achieving exactly-once semantics.

For more details on the incremental snapshot algorithm, see MySQL CDC Connector.

For tables with a single-column primary key, chunks are split based on that key by default. For tables with a composite primary key, the first column of the primary key is used for splitting by default. Ververica Runtime (VVR) 6.0.7 and later supports reading source tables without a primary key. You must set the scan.incremental.snapshot.chunk.key-column parameter to specify a non-nullable column for splitting.

Accelerate reads in the snapshot phase

During the snapshot phase, the MySQL source table reads snapshot data through a JDBC connection. Use the following methods to accelerate reads during this phase.

1. Increase the source parallelism to speed up reads during the snapshot phase.

2. Increase the value of scan.incremental.snapshot.chunk.size to fetch more data in a single chunk.

3. If the downstream result table has a primary key and supports idempotent writes, you can enable scan.incremental.snapshot.backfill.skip to skip reading the binary log for the backfill part. This speeds up processing during the snapshot phase.

Enable source reuse to reduce binlog connections

When a job includes multiple MySQL source tables, you can enable source reuse to reduce the database load by sharing a single binary log 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.

Enable the source reuse feature in an SQL job by using the SET command:

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

We recommend enabling source reuse only for new jobs. If you enable source reuse for an existing job, you must perform a stateless restart. This is because source reuse changes the job topology, and starting from an existing state may cause the job to fail or result in data loss.

After you enable source reuse, MySQL source tables with the same configuration parameters are merged. If all source tables in your job share the same configuration, the number of binary log connections is calculated as follows:

• During the snapshot phase, the number of binary log connections is equal to the source parallelism.

• During the incremental phase, the number of binary log connections is 1.

Enable binlog parsing to accelerate incremental reads

When used as a data source or for data ingestion, the MySQL connector parses binary log files to generate change messages during the incremental phase. The binary log file records all table changes in binary format. Use the following methods to accelerate binary log file parsing.

• 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 the event stream from the database, it places events in a blocking queue before they are passed 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: Specifies how long, in milliseconds, the connector waits before requesting new change events. The default value is 1,000 milliseconds, or 1 second.

Example:

CREATE TABLE mysql_source (...) WITH (
    'connector' = 'mysql-cdc',
    -- Debezium configurations
    'debezium.max.queue.size' = '162580',
    'debezium.max.batch.size' = '40960',
    'debezium.poll.interval.ms' = '50',
    -- Enable parallel parsing and parsing filters
    'scan.only.deserialize.captured.tables.changelog.enabled' = 'true',  -- Parse change events only for specified tables.
    'scan.parallel-deserialize-changelog.enabled' = 'true'  -- Use multiple threads to parse the binary log.
    ...
)

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:

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',  // The ID of the database instance.
    'rds.main-db-id' = '12345678', // The ID of the primary database.
    '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.