Realtime Compute for Apache Flink:MySQL

Background information

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

Features

A MySQL Change Data Capture (CDC) source table first reads the full historical data of a database and then seamlessly switches to reading binary log (binlog) events. This process ensures exactly-once semantics, which means no data is missed or duplicated, even if failures occur. The MySQL CDC source table supports concurrent reading of full data and implements lock-free reading and resumable data transfer using an incremental snapshot algorithm. For more information, see About MySQL CDC source tables.

Prerequisites

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

Limits

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 reuse connections, binlog offsets become disordered. This results in over-read or under-read data.

  Server ID configuration in different scenarios

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

  • Parallelism = 1 or incremental snapshot disabled

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

  • Parallelism > 1 and incremental snapshot enabled

    ## Specify a Server ID range. Ensure the number of available Server IDs in the range is at least equal to the parallelism. Assume parallelism is 3.
    SELECT * FROM source_table /*+ OPTIONS('server-id'='123456-123458') */ ;

  • Data synchronization with CTAS

    When you use CTAS for data synchronization and the CDC data sources have identical configurations, 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. In this case, you must assign a different Server ID to 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 it writes data.

  • You must declare at least one non-primary key field. Otherwise, an error occurs.

  • NOT ENFORCED in the DDL statement indicates that Flink does not enforce validity checks 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.

  The Flink-generated SQL may be rewritten by the optimizer, which prevents the index from being used during actual database queries. To verify whether an index is used, you can check the execution plan (EXPLAIN) or slow query log in MySQL to view 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

About MySQL CDC source tables

• How it works

  A MySQL CDC source table scans the full table at startup and splits it into multiple chunks based on the primary key. It records the current binlog offset and then uses the incremental snapshot algorithm to read each chunk's data using SELECT statements. The job performs periodic checkpoints to record completed chunks. If a failover occurs, the job continues to read only unfinished chunks. After all chunks are read, the job reads incremental change records from the previously recorded binlog offset. The Flink job continues to perform periodic checkpoints to record the binlog offset. If a failover occurs, the job resumes processing from the last recorded binlog offset. This process achieves exactly-once semantics.

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

• Metadata

  Metadata is highly useful in sharded database and table merging scenarios. 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 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 containing the row.
  table_name	STRING NOT NULL	The name of the table containing the row.
  op_ts	TIMESTAMP_LTZ(3) NOT NULL	The time when the row was changed in the database. If the record comes from the table's existing historical data rather than 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	Read the MySQL query log record corresponding to this row. Note MySQL must have the binlog_rows_query_log_events parameter enabled to record query logs.

  The following example shows how to merge multiple orders tables from different sharded databases in a MySQL instance and synchronize them to the 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 preceding code, if you set scan.read-changelog-as-append-only.enabled to true in the WITH clause, the output varies based on the primary key configuration of the downstream table:

  • If the primary key of the downstream table is order_id, the output contains only the last change for each primary key in the upstream table. For example, if the last change for a primary key is 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 order_id, operation_ts, and op_type, the output contains the complete change history for each primary key in the upstream 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 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 matches database names that start with test, such as test1 or test2.

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

  • txc is an exact match. It matches the specific database name txc.

  When matching fully qualified table names, 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) matches 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 commas (,) to specify multiple tables or databases.

  • To match multiple tables or use multiple regular expressions, you can separate them with a VERTICAL LINE (|) 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 LINE (|) operator. For example, the regular expression mytable_\d{1, 2} must be rewritten as the equivalent (mytable_\d{1}|mytable_\d{2}) to avoid using a comma.

• Concurrency control

  The MySQL connector supports the concurrent reading of full data, which improves data loading efficiency. When combined with Autopilot in the Realtime Compute for Apache Flink console, the connector automatically scales in during the incremental phase after the concurrent reading is complete. This saves 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 lets you set the parallelism for a specific VERTEX as needed.

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

  Important

  Regardless of whether you use Basic mode or Expert mode, the server-id range that is 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 eight unique server IDs. Therefore, the job can have a maximum of eight parallel tasks. Different jobs for the same MySQL instance must have non-overlapping server-id ranges. Each job must explicitly configure a different server-id.

• Autopilot Auto Scale-in

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

  Autopilot monitors traffic for each task in the MySQL CDC Source. When entering the binlog phase, if only one task handles binlog reading while others remain idle, Autopilot automatically reduces the Source's CU count and parallelism. To enable Autopilot, set the Autopilot mode to Active on the job's Operations and Maintenance page.

  Note

  The default minimum trigger interval for scaling down parallelism is 24 hours. For more information about Autopilot parameters 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 valid values are described as follows:

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

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

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

  • specific-offset: Skips the snapshot phase and starts reading from a specified binlog offset. You can specify the offset by binlog filename and position or by 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 filename 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 helps you restart the job from a specific checkpoint position.

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

• Keyless CDC source tables

  • To use a keyless table, you must set scan.incremental.snapshot.chunk.key-column and select only a non-null field.

  • The processing semantics of a keyless CDC source table depend on the behavior of the column that is specified in scan.incremental.snapshot.chunk.key-column:

    • If the specified column is never updated, exactly-once semantics are guaranteed.

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

• Read backup logs from RDS MySQL

  The MySQL CDC source table supports reading backup logs from Alibaba Cloud RDS MySQL. This feature is especially useful when the full snapshot phase takes a long time. In this case, local binlog files may be automatically cleaned up, while manually or automatically 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' = '60s'
      ...
  )

• Enable CDC source reuse

  A single job with multiple MySQL CDC source tables launches multiple binlog clients. If all source tables read from the same MySQL instance, this practice increases the pressure on the database. For more information, see MySQL CDC FAQ.

  Solutions

  VVR 8.0.7 and later versions support MySQL CDC source reuse. Reuse merges compatible MySQL CDC source tables. Merging occurs when source tables share identical configurations except for the database name, table name, and server-id. The engine automatically merges MySQL CDC sources within the same job.

  Procedure

  1. You can use the SET command in an SQL job:

     SET 'table.optimizer.source-merge.enabled' = 'true';
     
     # (VVR 8.0.8 and 8.0.9 versions) Also set this:
     SET 'sql-gateway.exec-plan.enabled' = 'false';

     VVR 11.1 and later versions enable reuse by default.

  2. Start the job statelessly. Modifying the source reuse configuration changes the job topology. You must start the job without states. Otherwise, the job may fail to start or lose data. If a source is merged, you can see a MergetableSourceScan node in the job topology.

  Important

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

  • In VVR 8.0.7, disabling operator chaining causes serialization issues.

Accelerate binlog reading

When you use the MySQL connector as a source table or data ingestion source, it parses binlog files to generate various change messages during the incremental phase. Binlog files record 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 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 events in a blocking queue before it writes them downstream. The default value is 8192.

  • debezium.max.batch.size: The maximum number of events that are processed per 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 ms (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',  -- Parse 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 consumes binlogs at a rate of 85 MB/s, which is about twice the rate of the open-source community version. If the binlog generation rate exceeds 85 MB/s (equivalent to one 512 MB file every 6 seconds), the Flink job latency continuously increases. The latency gradually decreases after the binlog generation rate slows down. When binlog files contain large transactions, the processing latency may temporarily increase and then 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 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 checkpointing.
    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 maintain 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 information about issues that you might encounter when you use a CDC source table, see CDC issues.