Realtime Compute for Apache Flink:Best practices for MySQL connector

Set the Server ID to avoid Binlog consumption conflicts

Each client that synchronizes database data has a unique server ID. If multiple MySQL CDC source tables share the same server ID, errors of server ID conflicts will occur, so we recommend assigning a different server ID for each client.

• Server ID configuration

  A server ID can be set in the Flink table's DDL statement or via SQL hints.

  We recommend that you assign the server ID via SQL hints. For more information about SQL hints, see SQL hints.

• Configuration of server IDs in different scenarios

  • Scenario 1: Incremental snapshot is disabled or the parallelism is 1.

    You can specify a Server ID if the incremental snapshot frame is disabled or the degree of parallelism is 1.

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

  • Scenario 2: Incremental snapshot is enabled and the parallelism is greater than 1.

    Specify a range of server IDs, and ensure that the number of available server IDs within the range is not less than the parallelism. For example, when the parallelism is 3, execute the following statement to set a server ID range:

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

  • Data synchronization using CTAS

    If you use the CTAS to synchronize data and multiple MySQL CDC sources share the same configurations, the source tables are automatically merged. In this case, you can specify the same server ID for multiple MySQL CDC source tables. For more information, see the "Example 4: execution of multiple CREATE TABLE AS statements" section of the CREATE TABLE AS statement topic.

  • Scenario 4: A job contains multiple MySQL CDC source tables, and the CTAS statement is not used for data synchronization.

    If a job contains multiple MySQL CDC source tables, is not synchronized using a CTAS statement, and has Source reuse disabled, you must provide a unique Server ID for each CDC source table (for more information, see Enable Source reuse to reduce Binlog data connections). Similarly, if the incremental snapshot framework is enabled and the degree of 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;

Configure chunk options for memory optimization

When the MySQL CDC source connector is started, the connector scans the whole table whose data needs to be read, splits the table into multiple chunks based on the primary key, and then records the binary log file position at this time. Then, the MySQL CDC source connector implements the incremental snapshot mechanism to read data from each chunk. The Flink job periodically generates checkpoints to record the chunks whose data is read. If a failover occurs, the MySQL CDC connector needs to only continue reading data from the chunks whose data is not read. After the data of all chunks is read, incremental change records are read from the previous binary log file position. The Flink job continues periodically generating checkpoints to record the binary log file position. If a failover occurs, the MySQL CDC connector processes data from the previous binary log file position. This way, the exactly-once semantics is implemented.

For more information about the incremental savepoint algorithm, see MySQL CDC connector in the Apache Flink documentation.

A table with a single-column primary key is split into chunks based on that key by default. A physical table with a composite primary key is chunked by the first column of the key by default. Realtime Compute for Apache Flink that uses Ververica Runtime (VVR) 6.0.7 or later supports reading data from a table without a primary key. Data in such a table is distributed into chunks based on the non-null column specified by scan.incremental.snapshot.chunk.key-column.

Adjust job configurations to accelerate reads in the full phase

During the full read phase, the MySQL source table reads snapshot data through a Java Database Connectivity (JDBC) connection. You can accelerate reads in the full phase in the following ways:

1. Increase the source concurrency to accelerate reads in the full phase.

2. Increase the value of scan.incremental.snapshot.chunk.size to increase the data volume retrieved by a single chunk.

3. If the downstream sink 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 accelerates processing in the full phase.

Enable Source Reuse to Reduce Binlog Data Connections

Source merging is useful for a job with multiple MySQL CDC source tables. It enables Flink to access the binary log via minimum connections necessary to MySQL, reducing the load on the MySQL database. This feature is only supported by the Realtime Compute for Apache Flink connector of MySQL CDC. The Apache Flink connector of MySQL CDC does not support this.

You can use the SET command in an SQL job to enable the Source reuse feature:

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

We recommend that you enable the Source reuse feature for newly created jobs. If you enable this feature for an existing job, you must perform a stateless start. This is because Source reuse changes the job topology. Starting the job from its previous state may cause a startup failure or data loss.

After source merging is enabled, MySQL CDC source tables with the same configurations are merged. If all MySQL CDC source tables share the same configurations, the number of MySQL connections in the corresponding Flink job is as follows:

• During snapshot reading, the number of connections equals the source parallelism.

• During incremental reading, the number of connections is 1.

Configure binlog parsing options to accelerate incremental reading

If the MySQL CDC connector is used as a source table or a data ingestion source, the MySQL CDC connector parses binary log files to generate change messages. Binary log files record the change data of all tables. You can use the following methods to accelerate the parsing of binary log data:

• Enable parallel parsing and parsing filter. This feature is supported only by the Realtime Compute for Apache Flink connector of MySQL CDC in VVR 8.0.7 or later. It is not supported by the Apache Flink connector of MySQL CDC.

  • Enable the scan.only.deserialize.captured.tables.changelog.enabled configuration item. This item parses only the change events of specified tables.

  • Enable the scan.parallel-deserialize-changelog.enabled configuration item. This item uses multiple threads to parse the binary log file and sequentially sends the parsed data to the consumption queue. When you enable this configuration, you may need to increase the Task Manager CPU.

• Optimize Debezium-related options

  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 an event stream from the database, it places the events in the blocking queue before it writes them downstream. The default value is 8192.

  • debezium.max.batch.size: The maximum number of events that the connector can process in each iteration. The default value is 2048.

  • debezium.poll.interval.ms: The number of milliseconds that the connector waits before it requests new change events. The default value is 1000 milliseconds, which is 1 second.

Example:

CREATE TABLE mysql_source (...) WITH (
    'connector' = 'mysql-cdc',
    -- Configure Debezium-related options.
    'debezium.max.queue.size' = '162580',
    'debezium.max.batch.size' = '40960',
    'debezium.poll.interval.ms' = '50',
    -- Enable parallel parsing and parsing filter configuration.
    'scan.only.deserialize.captured.tables.changelog.enabled' = 'true', -- Parses only the change events in the specified tables. 
    'scan.parallel-deserialize-changelog.enabled' = 'true' -- Uses multiple threads to parse events in binlog files. 
    ...
)

Enable reading of RDS binary logs to prevent expiration

You can use ApsaraDB RDS for MySQL as a data source. This enables you to read log backups stored in OSS. When the requested binary log file (specified by a timestamp or position) is stored in OSS, Flink downloads the binary log to its cluster before processing it. When the requested binary log file is available locally, Flink automatically switches to the database connection to read the binary log. This feature is only supported by the Realtime Compute for Apache Flink connector of MySQL CDC. The Apache Flink connector of MySQL CDC does not support this.

To read binary logs from OSS, configure ApsaraDB RDS for MySQL connection options as follows:

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 ApsaraDB RDS for MySQL instance ID. 
    'rds.main-db-id' = '12345678', -- The primary database ID. 
    'rds.endpoint' = 'rds.aliyuncs.com'
    ...
)

Synchronize database data and schema changes

For data synchronization tasks, we recommend creating a data ingestion deployment, which is optimized for data integration scenarios. For more information, see Getting started with a YAML deployment for data ingestion and Develop a Flink CDC job (Beta).

The following code snippet shows how a data ingestion deployment synchronizes data and schema changes from the MySQL database named 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

Add table feature for data ingestion connectors

The MySQL data ingestion connector provides configuration items for supporting new tables in two scenarios.