Realtime Compute for Apache Flink:MySQL connector

Background information

The MySQL connector supports all databases that are compatible with the MySQL protocol. The databases include ApsaraDB RDS for MySQL, PolarDB for MySQL, OceanBase database in MySQL-compatible mode, and self-managed MySQL databases.

Features

A MySQL Change Data Capture (CDC) source table is a streaming source table of MySQL databases. The MySQL CDC source connector reads full historical data from a MySQL database and then reads binary log data. This way, data accuracy is ensured. If an error occurs, the exactly-once semantics can be used to ensure data accuracy. The MySQL CDC source connector supports full data reading in multiple subtasks at the same time. You can use the incremental savepoint algorithm to perform lock-free reading and resumable uploads. For more information, see Basic concepts about MySQL CDC source tables.

Prerequisites

• CDC source table and data ingestion source

  You have performed the operations described in the Configure a MySQL database topic to meet the following requirements before you use a MySQL CDC source table:

  • A network connection is established between your MySQL database and Ververica Platform (VVP).

  • MySQL server configurations

    • The MySQL version is V5.6, V5.7, or V8.0.X.

    • Binary logging is enabled.

    • The format of binary logs is set to ROW.

    • The binlog_row_image parameter is set to FULL.

  • The interactive_timeout and wait_timeout parameters are configured in the MySQL configuration file.

  • A MySQL user is created and granted the SELECT, SHOW DATABASES, REPLICATION SLAVE, and REPLICATION CLIENT permissions.

• Dimension table and sink table

  • A MySQL database and a MySQL table are created. For more information, see Create a database and account for an ApsaraDB RDS for MySQL instance, Create a database and account for a PolarDB for MySQL instance, or Create a database and account for a self-managed MySQL instance.

  • An IP address whitelist is configured. For more information, see Configure an ApsaraDB RDS for MySQL whitelist, Configure a PolarDB for MySQL whitelist, or Configure a self-managed MySQL whitelist.

Limits

• CDC source table and data ingestion source

  • Only Realtime Compute for Apache Flink that uses Ververica Runtime (VVR) 4.0.8 or later supports lock-free reading and parallel reading.

  • You must select a VVR version based on the MySQL version. The following table describes the mappings between VVR and MySQL versions.

    You can run the select version() command to query the MySQL version.

    VVR version	Supported MySQL version
    VVR 4.0.8 to VVR 4.0.10	5.7
    	8.0.X
    VVR 4.0.11 and later	5.6.X
    	5.7.X
    	8.0.X

    Important

    The default value of the scan.incremental.snapshot.enabled parameter is true. This indicates that the incremental savepoint feature is enabled by default. This ensures that ApsaraDB RDS for MySQL 5.6.X runs as expected. You cannot disable the incremental savepoint feature for ApsaraDB RDS for MySQL 5.6.X. This limit is removed from ApsaraDB RDS for MySQL 6.0.8 and ApsaraDB RDS for MySQL 8.0.1. Therefore, you can disable the incremental savepoint feature for ApsaraDB RDS for MySQL 6.0.8 and ApsaraDB RDS for MySQL 8.0.1. We recommend that you do not disable the incremental savepoint feature. If you disable this feature, the MySQL database is locked and the online business processing performance may be affected.

  • MySQL CDC source tables do not support watermarks.

  • Only MySQL users who are granted specific permissions can use the MySQL CDC source connector to read full data and incremental data from a MySQL database. The permissions include SELECT, SHOW DATABASES, REPLICATION SLAVE, and REPLICATION CLIENT.

  • If you use the CREATE TABLE AS statement together with the CREATE DATABASE AS statement to perform data synchronization, specific schema changes of the MySQL CDC source table can be synchronized to the sink table. For more information about the supported change types, see Synchronization policies for table schema changes. In other scenarios, the schema changes of the MySQL CDC source table cannot be synchronized to the sink table.

  • MySQL CDC source tables do not support the synchronization of truncate operations.

  • We recommend that you do not read data from a secondary instance or a read-only secondary instance of ApsaraDB RDS for MySQL. By default, the binary log retention period of the secondary instance or read-only secondary instance of ApsaraDB RDS for MySQL is short. If the binary log file is deleted due to expiration, the deployment cannot consume the binary log data and an error occurs.

  • Data of Multi-master Cluster (Database/Table) Edition of PolarDB for MySQL of 1.0.19 or earlier cannot be read for the MySQL CDC source table. For more information about Multi-master Cluster (Database/Table) Edition, see Overview. Duplicate table IDs may be generated in binary log data of Multi-master Cluster (Database/Table) Edition of PolarDB for MySQL of 1.0.19 or earlier. This leads to incorrect mapping of the schema of the MySQL CDC source table. As a result, an error occurs when the binary log data is parsed. This issue is resolved in PolarDB for MySQL of a version later than 1.0.19.

• Dimension table and sink table

  • Only Realtime Compute for Apache Flink that uses VVR 4.0.11 or later supports the MySQL connector.

  • The at-least-once semantics can be used. If an ApsaraDB RDS for MySQL sink table contains a primary key, idempotence can be used to ensure data correctness.

Precautions

• CDC source table and data ingestion source

  • You must specify a unique server ID for each MySQL CDC data source.

    • Use of server IDs

      Each client that synchronizes data from a database has a unique server ID. The MySQL server maintains network connections and binary log file positions based on the server ID. If a large number of clients that have different server IDs connect to the MySQL server, the CPU utilization of the MySQL server may significantly increase. This affects the stability of online business.

      If multiple MySQL CDC data sources use the same server ID and the data sources cannot be merged as one source operator, the checkpoints for binary log positions may be out of order. As a result, the amount of data that is read may be greater than or less than expected. A conflict between server IDs may also occur. For more information, see Upstream and downstream storage. To prevent the preceding issue, we recommend that you specify a unique server ID for each MySQL CDC data source.

    • Configuration methods of server IDs

      You can specify server IDs in DDL statements or by using dynamic hints.

      We recommend that you use dynamic hints to configure server IDs instead of configuring server IDs in DDL statements. For more information about dynamic hints, see SQL hints.

    • Configuration of server IDs in different scenarios

      • Scenario 1: The incremental savepoint algorithm is disabled or the degree of parallelism is 1.

        If the incremental savepoint algorithm is disabled or the degree of parallelism is 1, you can specify a server ID.

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

      • Scenario 2: The incremental savepoint algorithm is enabled and the degree of parallelism is greater than 1.

        If the incremental savepoint algorithm is enabled and the degree of parallelism is greater than 1, you must specify a server ID range. Make sure that the number of available server IDs in the range is not less than the degree of parallelism. The following statement shows the configuration of server IDs when the degree of parallelism is 3.

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

      • Scenario 3: The CREATE TABLE AS statement is executed to synchronize data.

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

      • Scenario 4: Multiple MySQL CDC source tables are used in a deployment in which the CREATE TABLE AS statement is not executed.

        If multiple MySQL CDC source tables are used in a deployment and the CREATE TABLE AS statement is not executed, the data sources cannot be merged as one source operator. You must specify a unique server ID for each MySQL CDC source table. Similarly, if the incremental savepoint algorithm is enabled and the degree of parallelism is greater than 1, you must specify the 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;

  • Only deployments that use VVR 4.0.8 or later support lock-free reading, parallel reading, and resumable uploads during full data reading.

    If your deployment uses a VVR version that is earlier than 4.0.8, you must grant the RELOAD permission to the MySQL user to obtain the global read lock. This ensures data reading consistency. The global read lock may block write operations for a few seconds. This may affect online business.

    If your deployment uses a VVR version that is earlier than 4.0.8, checkpoints cannot be generated during full data reading. If your deployment fails in the process, the deployment reads full data again. This reduces the stability of data reading performance. Therefore, we recommend that you update the version of VVR that is used by your deployment to 4.0.8 or later.

• Sink table

  • ApsaraDB RDS for MySQL supports the auto-increment primary key. Therefore, you do not need to declare the auto-increment field in the DDL statement. For example, if you use ID as an auto-increment field, you do not need to declare the ID field in the DDL statement. When a row of output data is written to the ApsaraDB RDS for MySQL database, a value is automatically filled for the auto-increment field.

  • You must declare at least one non-primary key in the DDL statement. Otherwise, an error is returned.

  • NOT ENFORCED indicates that Flink does not perform mandatory verification on the primary key. You must ensure the correctness and integrity of the primary key.

    Realtime Compute for Apache Flink does not fully support mandatory verification. Realtime Compute for Apache Flink considers that the primary key is correct based on the assumption that the nullability of columns is aligned with the columns in the primary key. For more information, see Validity Check.

• Dimension table

  If you want to perform queries by index when the MySQL connector is used for a dimension table, you must follow the leftmost prefix matching principle of MySQL to sort the columns that are specified in the JOIN operation. However, this cannot ensure that indexes are used for queries. Specific filter conditions may be rewritten or modified due to SQL optimization. As a result, the filter conditions that are obtained by the connector may not hit indexes. To determine whether the connector queries data by index, check the specific SELECT statements that are executed on the related database.

Sample code

• CDC source table

  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>'
  );
  
  CREATE TEMPORARY TABLE blackhole_sink(
    order_id INT,
    customer_name STRING
  ) WITH (
    'connector' = 'blackhole'
  );
  
  INSERT INTO blackhole_sink
  SELECT order_id, customer_name FROM mysqlcdc_source;

• Dimension table

  CREATE TEMPORARY TABLE datagen_source(
    a INT,
    b BIGINT,
    c STRING,
    `proctime` AS PROCTIME()
  ) WITH (
    'connector' = 'datagen'
  );
  
  CREATE TEMPORARY TABLE mysql_dim (
    a INT,
    b VARCHAR,
    c VARCHAR
  ) WITH (
    'connector' = 'mysql',
    'hostname' = '<yourHostname>',
    'port' = '3306',
    'username' = '<yourUsername>',
    'password' = '<yourPassword>',
    'database-name' = '<yourDatabaseName>',
    'table-name' = '<yourTableName>'
  );
  
  CREATE TEMPORARY TABLE blackhole_sink(
    a INT,
    b STRING
  ) WITH (
    'connector' = 'blackhole'
  );
  
  INSERT INTO blackhole_sink
  SELECT T.a, H.b
  FROM datagen_source AS T JOIN mysql_dim FOR SYSTEM_TIME AS OF T.`proctime` AS H ON T.a = H.a;

• Sink table

  CREATE TEMPORARY TABLE datagen_source (
    `name` VARCHAR,
    `age` INT
  ) WITH (
    'connector' = 'datagen'
  );
  
  CREATE TEMPORARY TABLE mysql_sink (
    `name` VARCHAR,
    `age` INT
  ) WITH (
    'connector' = 'mysql',
    'hostname' = '<yourHostname>',
    'port' = '3306',
    'username' = '<yourUsername>',
    'password' = '<yourPassword>',
    'database-name' = '<yourDatabaseName>',
    'table-name' = '<yourTableName>'
  );
  
  INSERT INTO mysql_sink
  SELECT * FROM datagen_source;

• Data ingestion source

  source:
    type: mysql
    name: MySQL Source
    hostname: ${mysql.hostname}
    port: ${mysql.port}
    username: ${mysql.username}
    password: ${mysql.password}
    tables: ${mysql.source.table}
    server-id: 7601-7604
  
  sink:
    type: values
    name: Values Sink
    print.enabled: true
    sink.print.logger: true

Basic concepts about MySQL CDC source tables

• Principles

  When the MySQL CDC source table is started, it scans the whole table, divides the table into multiple shards (chunks) based on the primary key, and records the binary log location at this time. Then, the MySQL CDC source connector uses the incremental savepoint algorithm to read data from each chunk by using the SELECT statement. The deployment periodically generates checkpoints to record the chunks whose data is read. If a failover occurs, the MySQL CDC source 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 previously obtained binary log file position. The deployment continues periodically generating checkpoints to record the binary log file position. If a failover occurs, the MySQL CDC source 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.

• Metadata

  In most cases, access to metadata is required when you merge and synchronize tables in a sharded database. If you expect to identify data records by the source database names and table names after tables are merged, you can configure metadata columns in the data merging statement to read the source database name and table name of each data record. This way, you can identify the source of each data record after tables are merged.

  MySQL CDC source tables that are used by Realtime Compute for Apache Flink of vvr-4.0.11-flink-1.13 or later support the metadata column syntax. The following table describes the metadata that you can access by using metadata columns.

  Metadata key	Metadata type	Description
  database_name	STRING NOT NULL	The name of the source database to which the current data record belongs.
  table_name	STRING NOT NULL	The name of the source table to which the current data record belongs.
  op_ts	TIMESTAMP_LTZ(3) NOT NULL	The time when the current data record changes in the database. If the data record is obtained from the historical data of the table instead of from the binary log file, the value of the metadata key is fixed to 0.
  op_type	STRING NOT NULL	The change type of the current data record. Valid values: +I: indicates an INSERT message. -D: indicates a DELETE message. -U: indicates an UPDATE_BEFORE message. +U: indicates an UPDATE_AFTER message. Note Only Realtime Compute for Apache Flink that uses VVR 8.0.7 or later supports this parameter.

  The following sample code shows how to merge multiple orders tables in database shards of a MySQL instance into a MySQL table named mysql_orders and synchronize data from the MySQL table to a Hologres table named holo_orders.

  CREATE TABLE mysql_orders (
    db_name STRING METADATA FROM 'database_name' VIRTUAL,  -- Read the database name. 
    table_name STRING METADATA  FROM 'table_name' VIRTUAL, -- Read the table name. 
    operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL, -- Read the change time. 
    op_type STRING METADATA FROM 'op_type' VIRTUAL, -- Read the change type. 
    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-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'username' = 'flinkuser',
    'password' = 'flinkpw',
    'database-name' = 'mydb_.*', -- Use a regular expression to match multiple database shards. 
    'table-name' = 'orders_.*'   -- Use a regular expression to match multiple tables in the sharded database. 
  );
  
  INSERT INTO holo_orders SELECT * FROM mysql_orders;

  If you add the 'scan.read-changelog-as-append-only.enabled' = 'true' configuration to the WITH clause in the preceding sample code, the output of the sample code varies based on the primary key of the downstream table:

  • If the primary key of the downstream table is order_id, the output contains only the last change data of each primary key value of the upstream table. If the last change of a primary key value of the upstream table is a delete operation, a data record with the same primary key value and op_type as -D is contained in the downstream table.

  • If the primary key of the downstream table is composed of order_id, operation_ts, and op_type, the output contains the complete change data of each primary key value of the upstream table.

• Regular expressions

  MySQL CDC source tables allow you to use regular expressions in table names or database names to match multiple tables or databases. The following sample code provides an example on how to use a regular expression to specify multiple tables.

  CREATE TABLE products (
    db_name STRING METADATA FROM 'database_name' VIRTUAL,
    table_name STRING METADATA  FROM 'table_name' VIRTUAL,
    operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
    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-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'username' = 'root',
    'password' = '123456',
    'database-name' = '(^(test).*|^(tpc).*|txc|.*[p$]|t{2})', -- Use the regular expression to match multiple databases. 
    'table-name' = '(t[5-8]|tt)' -- Use the regular expression to match multiple tables. 
  );

  Elements in the preceding regular expressions:

  • In this example, ^(test).* is used to match a database whose name starts with the specified prefix. This expression can be used to match databases whose names start with test, such as test1 or test2.

  • In this example, .*[p$] is used to match a database whose name ends with the specified suffix. This expression can be used to match databases whose names end with p, such as cdcp or edcp.

  • In this example, txc is used to match the specified database name. This expression can be used to match a database that has a specific name, such as txc.

  When the MySQL CDC source connector matches a full-path table name, the connector uses the database name and table name to uniquely identify a table. This way, database-name.table-name is used as the pattern to match a table. For example, the (^(test).*| ^(tpc).*| txc |.*[p$]| t{2}).(t[ 5-8]| tt) pattern can match the txc.tt and test2.test5 tables in the database.

  Important

  In an SQL deployment, you cannot separate values of table-name and database-name with commas (,) to specify multiple tables or databases.

  • If you want to match multiple tables or use multiple regular expressions, you can use vertical bars (|) to connect the tables or regular expressions and enclose the tables or regular expressions in parentheses (). For example, if you want to read data from the user and product tables, you can set table-name to (user|product).

  • If you use a regular expression that contains commas (,), you must rewrite the regular expression into an equivalent regular expression that contains vertical bars (|). For example, if you use the mytable_\d{1, 2) regular expression, you must rewrite the regular expression into an equivalent regular expression (mytable_\d{1}|mytable_\d{2}) to prevent comma conflicts.

• Parallelism control

  The MySQL CDC source connector can read full data in multiple subtasks at the same time. This improves the efficiency of data loading. If you use the MySQL CDC source connector together with the Autopilot feature that is provided by the VVP of Realtime Compute for Apache Flink, automatic scale-in can be performed during incremental data reading after parallel reading is complete. This helps reduce the computing resources that are consumed.

  You can configure the Parallelism parameter in Basic mode or Expert mode in the Deployment Starting Configuration - Streaming dialog box in the development console of Realtime Compute for Apache Flink. The setting of the Parallelism parameter varies based on the resource configuration mode.

  • In Basic mode, the Parallelism parameter specifies the global parallelism of all deployments.

  • In Expert mode, you can configure the Parallelism parameter for a specific vertex node based on your business requirements.

  For more information about resource configurations, see Configure a deployment.

  Important

  When you configure the Parallelism parameter for a deployment in Basic mode or Expert mode, make sure that the range of the server-id parameter declared in the table is greater than or equal to the value of the Parallelism parameter for the deployment. For example, if the range of the server-id parameter is 5404-5412, eight unique server IDs can be used. Therefore, you can configure a maximum of eight parallel subtasks. In addition, the range specified by the server-id parameter for the same MySQL instance in different deployments cannot overlap. Each deployment must be explicitly configured with a unique server ID.

• Automatic scale-in by using Autopilot

  When full data is read, a large amount of historical data is accumulated. In most cases, Realtime Compute for Apache Flink reads historical data in parallel to improve the efficiency of data reading. When incremental data is read, only a single subtask is run because the amount of binary log data is small and the global order must be ensured. The numbers of compute units (CUs) that are required during full data reading and incremental data reading are different. You can use the Autopilot feature to balance performance and resource consumption.

  Autopilot monitors the traffic for each task that is used by the MySQL CDC source table. If the binary log data is read in only one task and other tasks are idle during incremental data reading, Autopilot automatically reduces the number of CUs and the degree of parallelism. To enable Autopilot, you need to only set the Mode parameter of Autopilot to Active on the Deployments page.

  Note

  By default, the minimum interval at which the degree of parallelism is decreased is 24 hours. For more information about automatic tuning and related parameters, see Configure automatic tuning.

• Startup mode

  You can configure the scan.startup.mode parameter to specify the mode in which the MySQL CDC source connector starts to read data from a MySQL database. Valid values:

  • initial: When the MySQL CDC source connector starts for the first time, the connector reads full data in a database table. After the full data reading is complete, the connector switches to the incremental reading mode to read binary log data. This is the default value.

  • earliest-offset: The MySQL CDC source connector skips the savepoint reading phase and starts to read accessible binary log data from the earliest offset.

  • latest-offset: The MySQL CDC source connector skips the savepoint reading phase and starts to read binary log data from the most recent offset. In this mode, the MySQL CDC source connector can read only data changes after the deployment is started.

  • specific-offset: The MySQL CDC source connector skips the savepoint reading phase and starts to read binary log data from the specified offset. You can specify the startup offset by using the name of a binary log file and the binary log file position or by using a GTID set.

  • timestamp: The MySQL CDC source connector skips the savepoint reading phase and starts to read binary log events from the specified timestamp.

  Sample code:

  CREATE TABLE mysql_source (...) WITH (
      'connector' = 'mysql-cdc',
      'scan.startup.mode'='earliest-offset, ' -- Starts to read data from the earliest offset. 
      'scan.startup.mode' = 'latest-offset', -- Starts to read data from the most recent offset. 
      'scan.startup.mode' = 'specific-offset', -- Starts to read data from the specified offset. 
      'scan.startup.mode' = 'timestamp', -- Starts to read data from the specified timestamp. 
      'scan.startup.specific-offset.file' = 'mysql-bin.000003', -- Specifies the name of the binlog log file when the scan.startup.mode parameter is set to specific-offset. 
      'scan.startup.specific-offset.pos' = '4', -- Specifies the binary log file position when the scan.startup.mode parameter is set to specific-offset. 
      'scan.startup.specific-offset.gtid-set' = '24DA167-0C0C-11E8-8442-00059A3C7B00:1-19', -- Specifies the GTID set when the scan.startup.mode parameter is set to specific-offset. 
      'scan.startup.timestamp-millis' = '1667232000000' -- Specifies the startup timestamp when the scan.startup.mode parameter is set to timestamp. 
      ...
  )

  Important

  • When a checkpoint is created, the MySQL CDC source connector records the current binary log event at the current offset as a log entry of the INFO level. The prefix of the log entry is Binlog offset on checkpoint {checkpoint-id}. The log entry can help you start the deployment from a specific checkpoint.

  • If the schema of a table is changed after the table is read, an error may occur when the scan.startup.mode parameter is set to earliest-offset, specific-offset, or timestamp. This is because the Debezium reader saves the latest table schema and earlier data that does not match the table schema cannot be correctly parsed.

• MySQL CDC source tables that do not have a primary key

  • In Realtime Compute for Apache Flink that uses VVR 6.0.7 or later, you can use MySQL CDC source tables that do not have a primary key. To use a MySQL CDC source table that does not have a primary key, you must configure the scan.incremental.snapshot.chunk.key-column parameter and specify only non-null fields.

  • The processing semantics of a MySQL CDC source table that does not have a primary key is determined based on the behavior of the columns that are specified by the scan.incremental.snapshot.chunk.key-column parameter.

    • If no update operation is performed on the specified columns, the exactly-once semantics is ensured.

    • If the update operation is performed on the specified columns, only the at-least-once semantics is ensured. However, you can specify the downstream primary key and perform the idempotence operation to ensure data correctness.

• Reading of backup binary log files from ApsaraDB RDS for MySQL

  The MySQL CDC source connector supports the reading of backup binary log files from ApsaraDB RDS for MySQL. This operation is suitable for scenarios in which full data reading takes a long period of time, local binary log files are automatically deleted, and the backup binary log files that are automatically or manually uploaded are available.

  Sample code:

  CREATE TABLE mysql_source (...) WITH (
      'connector' = 'mysql-cdc',
      'rds.region-id' = 'cn-beijing',
      'rds.access-key-id' = 'xxxxxxxxx', 
      'rds.access-key-secret' = 'xxxxxxxxx', 
      'rds.db-instance-id' = 'rm-xxxxxxxxxxxxxxxxx', 
      'rds.main-db-id' = '12345678',
      'rds.download.timeout' = '60s'
      ...
  )

• Enabling of the reuse of a MySQL CDC source table

  If multiple MySQL CDC source tables are used in the same deployment, the binlog client of each MySQL CDC source table is started. If a large number of MySQL CDC source tables are used and the MySQL tables to be read are in the same instance, the load on the database is high. For more information, see FAQ about CDC.

  Realtime Compute for Apache Flink that uses VVR 8.0.7 or later supports the reuse of a MySQL CDC source table. If the configurations, except the database name, table name, and server-id, of a MySQL CDC source table are the same as the configurations of another MySQL CDC source table, you can merge the source tables into one and reuse the merged source table. After you enable the reuse of MySQL CDC source tables, Realtime Compute for Apache Flink merges all MySQL CDC source tables in the same deployment that meet the merging requirements.

  You can run the following SET command to enable the reuse of a MySQL CDC source table in an SQL deployment:

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

  If you enable the reuse of a MySQL CDC source table in an existing SQL deployment, you must start the deployment without states. If you enable the reuse of a MySQL CDC source table in an existing SQL deployment, the deployment topology is changed. In this case, the deployment may not be started from the latest state, or data loss may occur.

  Important

  • In Realtime Compute for Apache Flink that uses VVR 8.0.8 and VVR 8.0.9, you must specify SET 'sql-gateway.exec-plan.enabled' = 'false' when you enable the reuse of a MySQL CDC source table in an SQL deployment.

  • After the operator chain is disconnected, the data writing from the MySQL CDC source table to the downstream operator increases the overhead of serialization and deserialization. Therefore, we recommend that you do not set the pipeline.operator-chaining parameter to false after you enable the reuse of a MySQL CDC source table. The more MySQL CDC source tables are merged, the higher the overhead is generated.

  • In Realtime Compute for Apache Flink that uses VVR 8.0.7, serialization issues may occur if you set the pipeline.operator-chaining parameter to false.

Binary log reading acceleration

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 configure the following parameters to accelerate the parsing of binary log data:

• Enable parallel parsing and parsing filter configuration

  • scan.only.deserialize.captured.tables.changelog.enabled: Parses only the change events in the specified tables.

  • scan.parallel-deserialize-changelog.enabled: Uses multiple threads to parse events in binary log files and send the parsed events to the consumption queue based on the event order in the binary log files.

• Optimize Debezium-related 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 can be used by blocking queues. When the Debezium reader reads an event stream from the database, the event stream is added to a blocking queue before the event stream is written to the downstream storage. Default value: 8192.

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

  • debezium.poll.interval.ms: The duration that the connector must wait before a new change event is requested. Default value: 1000. Unit: milliseconds.

Sample code:

CREATE TABLE mysql_source (...) WITH (
    'connector' = 'mysql-cdc',
    -- Configure Debezium-related parameters.
    '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', -- Parse only the change events in the specified tables. 
    'scan.parallel-deserialize-changelog.enabled' = 'true' -- Use multiple threads to parse events in binary log files. 
    ...
)

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: ${mysql.source.table}
  server-id: 7601-7604
  # Configure Debezium-related parameters.
  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
  scan.parallel-deserialize-changelog.enabled: true

MySQL CDC Enterprise Edition consumes binary log data at a rate of 85 MB/s, which is about twice that of the open source community. When the binary log file is generated at a speed that is greater than 85 MB/s, the processing latency of the Realtime Compute for Apache Flink deployment continues to increase. The processing latency of the deployment gradually decreases when the speed at which the binary log file is generated decreases. If a binary log file contains a large transaction, the processing latency may increase for a short period of time. After the transaction log is read, the processing latency may decrease.

MySQL CDC DataStream API

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
public class MySqlSourceExample {
  public static void main(String[] args) throws Exception {
    MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
        .hostname("yourHostname")
        .port(yourPort)
        .databaseList("yourDatabaseName") // set captured database
        .tableList("yourDatabaseName.yourTableName") // set captured table
        .username("yourUsername")
        .password("yourPassword")
        .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
        .build();
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // enable checkpoint
    env.enableCheckpointing(3000);
    env
      .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
      // set 4 parallel source tasks
      .setParallelism(4)
      .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering
    env.execute("Print MySQL Snapshot + Binlog");
  }
}

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-core</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-base</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-common</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-api-java-bridge</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>ververica-connector-mysql</artifactId>
    <version>${vvr.version}</version>
</dependency>

FAQ

For more information about issues that may occur when you use a MySQL CDC source table, see FAQ about CDC.