Realtime Compute for Apache Flink:MySQL connector

Overview

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

Features

The MySQL CDC source first takes a consistent snapshot of the existing data in a database and then seamlessly switches to reading the binary log (binlog) for change events. This process guarantees exactly-once semantics, ensuring no data is missed or duplicated, even during failures. The MySQL CDC source table supports concurrent reading of full data and implements lock-free reading and resumable data transfer by using an incremental snapshot algorithm. For more information, see About MySQL CDC source tables.

Prerequisites

Before you use a MySQL CDC source table, configure your MySQL database as described in Configure MySQL. These configurations are required to use a MySQL CDC source table.

Limitations

Notes

• Explicitly configure a different Server ID for each MySQL CDC data source

  Purpose of server ID

  You must explicitly configure a different Server ID for each MySQL CDC data source. If multiple MySQL CDC data sources share the same Server ID and cannot be reused, it causes disordered binlog offsets and can cause data to be over-read or under-read.

  Configure server ID in different scenarios

  You can specify the Server ID in the DDL statement, but we recommend that you configure it by using dynamic hints instead.

  • Parallelism = 1 or incremental snapshot disabled

    ## When the incremental snapshot framework is not enabled or the parallelism is 1, you can specify a specific Server ID.
    SELECT * FROM source_table /*+ OPTIONS('server-id'='123456') */ ;

  • Parallelism > 1 and incremental snapshot enabled

    ## You must specify a Server ID range. The number of available Server IDs in the range must be greater than or equal to the parallelism. Assume the parallelism is 3.
    SELECT * FROM source_table /*+ OPTIONS('server-id'='123456-123458') */ ;

  • Data synchronization with CTAS

    When you use CTAS for data synchronization, if the CDC data sources have the same configuration, the sources are automatically reused. In this case, you can configure the same Server ID for multiple CDC data sources. For more information, see Example 4: Multiple CTAS statements.

  • Multiple non-CTAS source tables that cannot be reused

    If a job contains multiple MySQL CDC source tables and does not use CTAS statements for synchronization, the data sources cannot be reused. You must provide a different Server ID for each CDC source table. Similarly, if the incremental snapshot framework 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;

• Sink table

  • Do not declare an auto-increment primary key in the DDL statement. MySQL automatically populates this field when writing data.

  • Declare at least one non-primary key field, otherwise an error is reported.

  • NOT ENFORCED in the DDL statement indicates that Flink does not perform a validity check on the primary key. You must ensure the correctness and integrity of the primary key. For more information, see Validity Check.

• Dimension table

  To use indexes for query acceleration, the field order in the JOIN clause must match the index definition order. This is known as the leftmost prefix rule. For example, if the index is on (a, b, c), the JOIN condition should be ON t.a = x AND t.b = y.

  Flink-generated SQL may be rewritten by the optimizer, which can prevent the index from being used during the actual database query. To verify whether an index is used, check the execution plan (EXPLAIN) or the slow query log in MySQL to see the actual SELECT statement that is executed.

Examples

• 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

MySQL CDC source tables

• How it works

  When the MySQL CDC source table starts, it scans the entire table and splits it into multiple chunks based on the primary key, recording the current binlog position. It then uses the incremental snapshot algorithm to read the data from each chunk one by one by using SELECT statements. The job periodically performs checkpoints to record the completed chunks. If a failover occurs, the job only needs to continue reading the unfinished chunks. After all chunks are read, it starts reading incremental change records from the previously recorded binlog position. The Flink job continues to perform periodic checkpoints to record the binlog position. If the job fails over, it resumes processing from the last recorded binlog position, thus achieving exactly-once semantics.

  For a more detailed explanation of the incremental snapshot algorithm, see MySQL CDC Connector.

• Metadata

  Metadata is highly useful in scenarios where you merge sharded databases and tables. After merging, businesses often still need to identify the source database and table for each row of data. Metadata columns allow you to access this information. Therefore, you can easily merge multiple sharded tables into a single destination table by using metadata columns.

  The MySQL CDC Source supports metadata column syntax. You can access the following metadata through metadata columns.

  Metadata key	Metadata type	Description
  database_name	STRING NOT NULL	The name of the database that contains the row.
  table_name	STRING NOT NULL	The name of the table that contains the row.
  op_ts	TIMESTAMP_LTZ(3) NOT NULL	The time when the row was changed in the database. If the record is from the table's existing historical data instead of the Binlog, this value is always 0.
  op_type	STRING NOT NULL	The change type of the row. +I: INSERT message -D: DELETE message -U: UPDATE_BEFORE message +U: UPDATE_AFTER message Note Supported only in Ververica Runtime (VVR) 8.0.7 and later.
  query_log	STRING NOT NULL	The MySQL query log record corresponding to the row read. Note To log queries, MySQL requires enabling binlog_rows_query_log_events.

  The following example shows how to merge multiple orders tables from different sharded databases in a MySQL instance and synchronize them to a holo_orders table in Hologres.

  CREATE TEMPORARY 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 timestamp.
    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_.*', -- Match multiple sharded databases using a regular expression.
    'table-name' = 'orders_.*'   -- Match multiple sharded tables using a regular expression.
  );
  
  INSERT INTO holo_orders SELECT * FROM mysql_orders;

  Based on the code above, if you set the scan.read-changelog-as-append-only.enabled option to true in the WITH clause, the output varies based on the primary key settings of the downstream table:

  • If the primary key of the downstream table is order_id, the output includes only the last change for each primary key in the source table. For example, if the last change for a primary key was a delete operation, you see a record in the downstream table with the same primary key and an op_type of -D.

  • If the primary key of the downstream table is a composite of order_id, operation_ts, and op_type, the output includes the complete change history for each primary key in the source table.

• Regular expression support

  The MySQL CDC source table supports using regular expressions in the table name or database name to match multiple tables or databases. The following example shows how to specify multiple tables by using a regular expression.

  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})', -- Match multiple databases using a regular expression.
    'table-name' = '(t[5-8]|tt)' -- Match multiple tables using a regular expression.
  );

  Explanation of the regular expressions in the preceding example:

  • ^(test).* is a prefix match example. This expression can match database names that start with test, such as test1 or test2.

  • .*[p$] is a suffix match example. This expression can match database names that end with p, such as cdcp or edcp.

  • txc is a specific match. It can match a specific database name, such as txc.

  When matching a fully qualified table name, MySQL CDC uses both the database name and table name to uniquely identify a table. It uses the pattern database-name.table-name for matching. For example, the pattern (^(test).*|^(tpc).*|txc|.*[p$]|t{2}).(t[ 5-8]|tt) can match tables such as txc.tt and test2.test5 in the database.

  Important

  In SQL job configurations, the table-name and database-name options do not support using a comma (,) to specify multiple tables or databases.

  • To match multiple tables or use multiple regular expressions, connect them with a vertical bar (|) and enclose them in parentheses. For example, to read the user and product tables, you can configure table-name as (user|product).

  • If a regular expression contains a comma, you must rewrite it using the vertical bar (|) operator. For example, the regular expression mytable_\d{1, 2} needs to be rewritten as the equivalent (mytable_\d{1}|mytable_\d{2}) to avoid using a comma.

• Concurrency control

  The MySQL connector supports reading full data with multiple concurrencies, which improves data loading efficiency. When combined with Autopilot in the Realtime Compute for Apache Flink console, it can automatically scale in during the incremental phase after the multi-concurrency reading is complete, thereby saving computing resources.

  In the Realtime Compute development console, you can set the job's parallelism on the Resource Configuration page in either Basic mode or Expert mode. The differences are as follows:

  • The parallelism set in Basic mode is the global parallelism for the entire job.

  • Expert mode allows you to set the parallelism for a specific VERTEX as needed.

  For more information about resource configuration, see Configure a job deployment.

  Important

  Whether you use Basic mode or Expert mode, the server-id range declared in the table must be greater than or equal to the job's parallelism. For example, if the server-id range is 5404-5412, there are 8 unique server IDs, so the job can have a maximum of 8 concurrencies. Different jobs for the same MySQL instance must have non-overlapping server-id ranges, meaning each job must explicitly configure a different server-id.

• Autopilot auto-scaling

  The full data phase accumulates a large amount of historical data. To improve reading efficiency, historical data is typically read concurrently. In the incremental binlog phase, however, because the volume of binlog data is small and a global order must be maintained, a single concurrency is usually sufficient. Autopilot can automatically balance performance and resources to meet these different requirements of the full and incremental phases.

  Autopilot monitors the traffic of each task in the MySQL CDC Source. When the job enters the binlog phase, if only one task is responsible for reading the binlog and the other tasks are idle, Autopilot will automatically reduce the CU count and parallelism of the Source. To enable Autopilot, set the Autopilot mode to Active on the job's Operations and Maintenance page.

  Note

  The minimum trigger interval for reducing parallelism is 24 hours by default. For more information on Autopilot options and details, see Configure Autopilot.

• Startup modes

  You can use the scan.startup.mode option to specify the startup mode for the MySQL CDC source table. The options include:

  • initial (default): Performs a full read of the database table upon the first startup, and then switches to incremental mode to read the binlog.

  • earliest-offset: Skips the snapshot phase and starts reading from the earliest available binlog position.

  • latest-offset: Skips the snapshot phase and starts reading from the end of the binlog. In this mode, the source table can only read data changes that occur after the job starts.

  • specific-offset: Skips the snapshot phase and starts reading from a specified binlog position. The position can be specified by the binlog filename and position, or by using a GTID set.

  • timestamp: Skips the snapshot phase and starts reading binlog events from a specified timestamp.

  Example:

  CREATE TABLE mysql_source (...) WITH (
      'connector' = 'mysql-cdc',
      'scan.startup.mode' = 'earliest-offset', -- Start from the earliest offset.
      'scan.startup.mode' = 'latest-offset', -- Start from the latest offset.
      'scan.startup.mode' = 'specific-offset', -- Start from a specific offset.
      'scan.startup.mode' = 'timestamp', -- Start from a specific timestamp.
      'scan.startup.specific-offset.file' = 'mysql-bin.000003', -- Specify the Binlog file name for the specific-offset mode.
      'scan.startup.specific-offset.pos' = '4', -- Specify the Binlog position for the specific-offset mode.
      'scan.startup.specific-offset.gtid-set' = '24DA167-0C0C-11E8-8442-00059A3C7B00:1-19', -- Specify the GTID set for the specific-offset mode.
      'scan.startup.timestamp-millis' = '1667232000000' -- Specify the startup timestamp for the timestamp mode.
      ...
  )

  Important

  • The MySQL source logs the current position at the INFO level during a checkpoint. The log prefix is Binlog offset on checkpoint {checkpoint-id}. This log can help you restart the job from a specific checkpoint position.

  • If the schema of the table being read has changed in the past, starting from the earliest-offset, a specific-offset, or a timestamp may cause errors. This is because the Debezium reader internally stores the latest schema, and older data with a mismatched schema cannot be parsed correctly.

• Keyless CDC source tables

  • To use a keyless table, you must set scan.incremental.snapshot.chunk.key-column and can only choose a non-nullable column.

  • The processing semantics of a keyless CDC source table are determined by the behavior of the column specified in scan.incremental.snapshot.chunk.key-column:

    • If the specified column is not updated, exactly-once semantics can be guaranteed.

    • If the specified column is updated, only at-least-once semantics can be guaranteed. However, you can achieve data correctness by combining it with a downstream system, specifying a downstream primary key, and using idempotent operations.

• Read backup logs from ApsaraDB RDS for MySQL

  The MySQL CDC source table supports reading backup logs from Alibaba Cloud ApsaraDB RDS for MySQL. This is particularly useful in scenarios where the full snapshot phase takes a long time, causing local binlog files to be automatically cleaned up, while automatically or manually uploaded backup files still exist.

  Example:

  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' = '60 s'
      ...
  )

• Enable CDC source reuse

  When a single job has multiple MySQL CDC source tables, each source table starts a corresponding binlog client. If there are many source tables and they all read from the same MySQL instance, it can put significant pressure on the database. For more information, see MySQL CDC FAQ.

  VVR 8.0.7+ versions support MySQL CDC source reuse. Different CDC source tables can be merged if all their configuration options, except for the database, table name, and server-id, are identical. After enabling source reuse, VVR will merge as many compatible MySQL CDC source tables as possible within the same job.

  Procedure

  1. Enable the source reuse feature in your SQL job draft using the SET command:

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

  2. Start the job without states. After enabling source reuse for an existing job, you must perform a stateless restart. This is because source reuse changes the job topology, and restarting from the old job state may fail or lead to data loss. If source reuse occurs, a MergetableSourceScan operator will show.

  Important

  • After you enable CDC source reuse, do not set pipeline.operator-chaining to false. Disabling operator chaining adds serialization and deserialization overhead. The more sources are merged, the greater the overhead becomes.

  • In VVR 8.0.7, disabling operator chaining causes a serialization issue.

Accelerate binlog reading

When you use the MySQL connector as a source table or a data ingestion source, it parses binlog files to generate various change messages during the incremental phase. The binlog file records all table changes in a binary format. You can accelerate binlog file parsing in the following ways.

• Enable parsing filter configuration

  • Use the scan.only.deserialize.captured.tables.changelog.enabled option to parse change events only for the specified tables.

• Optimize Debezium 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 a blocking queue before writing them 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: The number of milliseconds the connector should wait before requesting new change events. The default value is 1000 milliseconds, or 1 second.

Example:

CREATE TABLE mysql_source (...) WITH (
    'connector' = 'mysql-cdc',
    -- Debezium configuration
    'debezium.max.queue.size' = '162580',
    'debezium.max.batch.size' = '40960',
    'debezium.poll.interval.ms' = '50',
    -- Enable parsing filter
    'scan.only.deserialize.captured.tables.changelog.enabled' = 'true',  -- Parses change events only for specified tables.
    ...
)

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
  # Debezium configuration
  debezium.max.queue.size: 162580
  debezium.max.batch.size: 40960
  debezium.poll.interval.ms: 50
  # Enable parsing filter
  scan.only.deserialize.captured.tables.changelog.enabled: true

The enterprise version of MySQL CDC has a binlog consumption capacity of 85 MB/s, which is about twice that of the open-source community version. If the binlog generation speed exceeds 85 MB/s (equivalent to one 512 MB file every 6 seconds), the Flink job's latency will continue to increase. The processing latency will gradually decrease after the binlog generation speed slows down. When the binlog file contains large transactions, processing latency may temporarily increase and will decrease after the transaction's log is read.

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 the database to capture.
        .tableList("yourDatabaseName.yourTableName") // Set the table to capture.
        .username("yourUsername")
        .password("yourPassword")
        .deserializer(new JsonDebeziumDeserializationSchema()) // Converts a SourceRecord to a JSON string.
        .build();
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // Enable checkpointing.
    env.enableCheckpointing(3000);
    env
      .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
      // Set 4 parallel source tasks.
      .setParallelism(4)
      .print().setParallelism(1); // Use a parallelism of 1 for the sink to maintain message order.
    env.execute("Print MySQL Snapshot + Binary Log");
  }
}

<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 information about issues you might encounter when using a CDC source table, see CDC issues.