Realtime Compute for Apache Flink:MySQL

Overview

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

Benefits

The MySQL connector provides a unified stream of database changes by performing an initial snapshot followed by a seamless transition to binlog consumption. This architecture ensures Exactly-Once processing, preventing data loss or duplication even in the event of job failures. For more information, see Understanding MySQL source. Features:

Prerequisites

Before you use a MySQL CDC source table, complete the steps in Configure a MySQL database.

Limitations

Usage notes

• Source: Specify a unique server ID for each source.

  Purpose of server ID

  Each MySQL CDC data source must have a unique server ID. If multiple data sources share the same server ID and cannot reuse connections, binary log offsets may become inconsistent, leading to missing or duplicated data.

  Server ID configuration for different scenarios

  You can specify the server ID in DDL statements, but we recommend using dynamic hints instead of DDL parameters.

  • Parallelism = 1 or incremental snapshots disabled

    ## Specify a server ID when incremental snapshots are disabled or degree of parallelism is 1.
    SELECT * FROM source_table /*+ OPTIONS('server-id'='123456') */ ;

  • Parallelism > 1 and incremental snapshots enabled

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

  • Data synchronization using CTAS

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

  • Multiple non-CTAS source tables that cannot be reused

    If your job contains multiple MySQL CDC source tables and does not use CTAS, data sources cannot be reused. Assign a unique server ID to each CDC source table. Similarly, if incremental snapshots are enabled and the parallelism is greater than 1, 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 tables

  • Do not declare auto-increment primary keys in the DDL. MySQL fills them automatically.

  • At least one non-primary-key field must be declared. Otherwise, an error occurs.

  • NOT ENFORCED means Flink does not validate primary keys. You must ensure primary key correctness. For more information, see Validity check.

• Dimension tables

  To accelerate queries using indexes, the JOIN condition fields must match the index definition order (leftmost prefix rule). For example, if the index is (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 index usage. To verify index usage, check the execution plan (EXPLAIN) or slow query logs in MySQL.

Examples

• 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 (lookup) 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;

• Flink CDC 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

Understanding MySQL source

• How it works

  The MySQL CDC source utilizes an incremental snapshot algorithm to provide a lockless, high-performance, and exactly-once streaming experience:

  1. Snapshot phase: The table is split into chunks based on the primary key. These are read sequentially in parallel. Checkpoints occur regularly, allowing the job to resume from specific chunks upon failure.

  2. Binlog phase: Once all chunks are processed, the connector transitions to reading binlogs from the offset recorded at the start of the snapshot phase.

  3. Failover: The job resumes from the last successful checkpoint (either a specific chunk or a binlog offset), ensuring exactly-once semantics.

     See also: MySQL CDC connector in Apache Flink documentation.

• Metadata columns

  Metadata is essential when merging sharded tables into a single sink. Use the METADATA FROM syntax to access metadata columns:

  Metadata key	Type	Description
  database_name	STRING NOT NULL	The source database name.
  table_name	STRING NOT NULL	The source table name.
  op_ts	TIMESTAMP_LTZ(3) NOT NULL	The change timestamp (0 if historical snapshot data).
  op_type	STRING NOT NULL	Change type: +I, -D, +U (after), -U (before). Note Requires VVR 8.0.7+.
  query_log	STRING NOT NULL	The raw MySQL query (Requires binlog_rows_query_log_events). Note MySQL must have the binlog_rows_query_log_events parameter enabled to record query logs.

  Example: Merging sharded tables a single holo_orders Hologres sink

  CREATE TEMPORARY TABLE mysql_orders (
    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, -- Read change time.
    op_type STRING METADATA FROM 'op_type' VIRTUAL, -- Read 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_.*', -- Regex match multiple sharded databases.
    'table-name' = 'orders_.*'   -- Regex match multiple sharded tables.
  );
  
  INSERT INTO holo_orders SELECT * FROM mysql_orders;

  Additionally, if you set scan.read-changelog-as-append-only.enabled to true, the output varies based on the sink table's primary key configuration:

  • If the sink table's primary key is order_id, the output contains only the last change for each primary key from the source table. For a primary key whose last change was a delete operation, the sink table shows a record with the same primary key and an op_type of -D.

  • If the sink table's primary key is order_id, operation_ts, and op_type, the output contains the complete change history for each primary key from the source table.

• Regular expression support

  The MySQL connector supports matching multiple tables or databases using regular expressions. Example:

  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})', -- Regex match multiple databases.
    'table-name' = '(t[5-8]|tt)' -- Regex match multiple tables.
  );

  Regular expression explanation:

  • ^(test).*: Prefix match. Matches database names starting with "test", such as test1 or test2.

  • .*[p$]: Suffix match. Matches database names ending with "p", such as cdcp or edcp.

  • txc: Exact match.

  The MySQL connector identifies tables using the fully qualified format: database-name.table-name.

  Important

  No comma support: The table-name and database-name options do not support comma-separated lists. Use the vertical bar (|) operator inside parentheses to specify multiple values.

  Escaping special characters: If a pattern contains a comma or other special regex characters (like {1,2}), rewrite them using the | operator to ensure compatibility.

• Resource management

  • Parallelism: The server-id range must be greater than or equal to the job parallelism (e.g., range 5404-5412 supports up to 8 concurrent tasks). Each job must use a unique, non-overlapping server-id range.

  • Autopilot: When enabled, Flink automatically scales down resources after the snapshot phase completes, as the incremental binlog phase typically requires less parallelism than the initial snapshot load.

  In the Development Console, set the job's parallelism in basic or expert mode. The difference is as follows:

  • Basic mode: Parallelism applies to the job.

  • Expert mode: Vertext-specific parallelism is supported.

  For details, see Configure job deployments.

• Startup modes

  Use the scan.startup.mode configuration item to specify the startup mode for the MySQL CDC source table. Valid values:

  • initial: Performs an initial snapshot of the table, then transitions to reading binary logs.

  • latest-offset: Skips the snapshot; starts reading changes from the current end of the binary log.

  • earliest-offset: Skips the snapshot; starts reading from the earliest available binary log.

  • specific-offset: Skips the snapshot; starts from a user-defined position:

    • File/Pos: Use scan.startup.specific-offset.file and scan.startup.specific-offset.pos.

    • GTID: Use scan.startup.specific-offset.gtid-set.

  • timestamp: Skips the snapshot; starts from a specific time (in ms) provided by scan.startup.timestamp-millis.

  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 binary log filename for specific-offset mode.
      'scan.startup.specific-offset.pos' = '4', -- Specify the binary log position for specific-offset mode.
      'scan.startup.specific-offset.gtid-set' = '24DA167-0C0C-11E8-8442-00059A3C7B00:1-19', -- Specify the GTID set for specific-offset mode.
      'scan.startup.timestamp-millis' = '1667232000000' -- Specify the startup timestamp for timestamp mode.
      ...
  )

  Important

  • The MySQL source prints the current offset at checkpoint time with INFO level logging. The log prefix is Binlog offset on checkpoint {checkpoint-id}. This log helps you start the job from a specific checkpoint offset.

  • If the table being read has undergone schema changes, starting from the earliest-offset, specific-offset, or timestamp modes may cause errors. This is because the Debezium reader internally maintains the latest table schema, and earlier data with mismatched schemas cannot be parsed correctly.

• MySQL source tables without PK

  If your source table lacks a primary key, you must manually specify a chunk key column using the scan.incremental.snapshot.chunk.key-column option.

  • The processing semantics depend on whether the column specified in scan.incremental.snapshot.chunk.key-column:

  • Processing semantics:

    • Immutable chunk key: If the selected column is never updated, the connector guarantees exactly-once semantics.

    • Mutable chunk key: If the column is updated, the connector provides At-Least-Once semantics.

  • Best practice for data integrity: To ensure data consistency when using At-Least-Once semantics, define a primary key in your downstream system and ensure the sink operation is idempotent (e.g., using UPSERT or REPLACE logic) to handle potential duplicates.

• Reading ApsaraDB RDS for MySQL backup logs

  The MySQL CDC source can read archived binary logs from Object Storage Service (OSS). This is particularly useful in two scenarios:

  • Slow snapshot phase: When a snapshot takes a long time and the MySQL instance purges its local binary logs before the snapshot phase completes.

  • Log retention: When local binary logs have been deleted, but the backup files uploaded to OSS are still available.

    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'
        ...
    )

• Source reuse

  In a single job, each MySQL source table starts its own binlog client. When all source tables connect to the same instance, this increases the database load. For more information, see Flink CDC FAQ.

  Solution

  VVR 8.0.7 and later supports MySQL CDC Source reuse. Source reuse merges CDC source tables with compatible configurations. Tables are merged when their configurations are identical except for the database name, table name, and server-id.

  Procedure

  1. Use the SET command in your SQL job:

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

     VVR 11.1 and later have reuse enabled by default.

  2. Start the job stateless. Changing source reuse configurations changes the job topology, so the job must start without state. Otherwise, the job may fail to start or lose data. If sources are merged, you will see a MergetableSourceScan node.

  Important

  • After you enable reuse, do not disable operator chaining. Setting pipeline.operator-chaining to false increases data serialization and deserialization overhead. The more Sources merged, the greater the overhead.

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

Accelerate binlog reading

When the MySQL connector is used as a source, it parses binary logs to generate change events during the incremental phase. Binlog files record all table changes in binary format. Accelerate binary log file parsing using the following methods:

1. Enable parsing filter configuration

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

2. Optimize Debezium options

   debezium.max.queue.size: 162580
   debezium.max.batch.size: 40960
   debezium.poll.interval.ms: 50

   • debezium.max.queue.size: Max records the blocking queue can hold. When Debezium reads an event stream from the database, it places events into a blocking queue before writing them downstream. The default value is 8192.

   • debezium.max.batch.size: Max events the connector processes in each iteration. The default value is 2048.

   • debezium.poll.interval.ms: The number of milliseconds the connector waits before requesting new change events. The default value is 1000 milliseconds (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',  -- Only parse change events 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 fully managed MySQL connector can consume binary logs at up to 85 MB/s, approximately twice the throughput of the Apache Flink connector for MySQL. If the binlog generation rate exceeds 85 MB/s (a 512 MB file every 6 seconds), job latency increases. Latency decreases when the generation rate drops. Large transactions may briefly increase processing latency.

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>

Troubleshooting & FAQ

See Flink CDC FAQ.