MySQL CDC Source Table

Function

The MySQL CDC source table, that is, the MySQL streaming source table, reads all historical data in the database first and then smoothly switches data read to the Binlog to ensure data integrity.

Prerequisites

• MySQL CDC requires MySQL 5.7 or 8.0.x.
• An enhanced datasource connection has been created for DLI to connect to the MySQL database, so that you can configure security group rules as required.
  • For details about how to create an enhanced datasource connection, see Enhanced Datasource Connections in the Data Lake Insight User Guide.
  • For details about how to configure security group rules, see Security Group in the Virtual Private Cloud User Guide.
• In Flink cross-source development scenarios, there is a risk of password leakage if datasource authentication information is directly configured. You are advised to use the datasource authentication provided by DLI.

  For details about datasource authentication, see Introduction to Datasource Authentication.

• Binlog is enabled for MySQL, and binlog_row_image is set to FULL.
• A MySQL user has been created and granted the SELECT, SHOW DATABASES, REPLICATION SLAVE, and REPLICATION CLIENT permissions.

Precautions

• When creating a Flink OpenSource SQL job, you need to set Flink Version to 1.12 on the Running Parameters tab of the job editing page, select Save Job Log, and set the OBS bucket for saving job logs.
• Each client that synchronizes database data has a unique ID, that is, the server ID. You are advised to configure a unique server ID for each MySQL CDC job in the same database.
  Main reasons are as follows:

  • The MySQL server maintains the network connection and Binlog location based on the ID. Therefore, if a large number of clients with the same server ID connect to the MySQL server, the CPU usage of the MySQL server may increase sharply, affecting the stability of online services.
  • If multiple jobs share the same server ID, Binlog locations will be disordered, making data read inaccurate. Therefore, you are advised to configure different server IDs for each MySQL CDC job.

• Watermarks cannot be defined for MySQL CDC source tables. For details about window aggregation, see FAQ.
• If you connect to a sink source that supports upsert, such as GaussDB(DWS) and MySQL, you need to define the primary key in the statement for creating the sink table. For details, see the printSink table creation statement in Example.
• When using the MySQL CDM source table, do not manually disable debezium.connect.keep.alive in the source table parameters. Make sure that debezium.connect.keep.alive is set to true (default value).

  If debezium.connect.keep.alive is manually disabled and there is a connection exception between the Binlog thread and the MySQL server, the Binlog thread will not attempt to reconnect automatically. This may result in the inability to properly pull binlog logs from the source.

Syntax

create table mySqlCdcSource (
  attr_name attr_type 
  (',' attr_name attr_type)* 
  (','PRIMARY KEY (attr_name, ...) NOT ENFORCED)
)
with (
  'connector' = 'mysql-cdc',   
  'hostname' = 'mysqlHostname',
  'username' = 'mysqlUsername',
  'password' = 'mysqlPassword',
  'database-name' = 'mysqlDatabaseName',
  'table-name' = 'mysqlTableName'
);

Parameters

Example

In this example, MySQL-CDC is used to read data from RDS for MySQL in real time and write the data to the Print result table. The procedure is as follows (MySQL 5.7.32 is used in this example):

1. Create an enhanced datasource connection in the VPC and subnet where MySQL locates, and bind the connection to the required Flink elastic resource pool. For details, see Enhanced Datasource Connections.
2. Set MySQL security groups and add inbound rules to allow access from the Flink queue. Test the connectivity using the MySQL address by referring to Testing Address Connectivity. If the connection is successful, the datasource is bound to the queue. Otherwise, the binding fails.
3. Create a table named cdc_order in database flink of the MySQL database.

   CREATE TABLE `flink`.`cdc_order` (
   	`order_id` VARCHAR(32) NOT NULL,
   	`order_channel` VARCHAR(32) NULL,
   	`order_time` VARCHAR(32) NULL,
   	`pay_amount` DOUBLE  NULL,
   	`real_pay` DOUBLE  NULL,
   	`pay_time` VARCHAR(32) NULL,
   	`user_id` VARCHAR(32) NULL,
   	`user_name` VARCHAR(32) NULL,
   	`area_id` VARCHAR(32) NULL,
   	PRIMARY KEY (`order_id`)
   )	ENGINE = InnoDB
   	DEFAULT CHARACTER SET = utf8mb4
   	COLLATE = utf8mb4_general_ci;

4. Create a Flink OpenSource SQL job. Enter the following job script and submit the job.
   When you create a job, set Flink Version to 1.12 on the Running Parameters tab. Select Save Job Log, and specify the OBS bucket for saving job logs. Change the values of the parameters in bold as needed in the following script.

   create table mysqlCdcSource(
     order_id string,
     order_channel string,
     order_time string,
     pay_amount double,
     real_pay double,
     pay_time string,
     user_id string,
     user_name string,
     area_id STRING
   ) with (
     'connector' = 'mysql-cdc',
     'hostname' = 'mysqlHostname',
     'username' = 'mysqlUsername',
     'password' = 'mysqlPassword',
     'database-name' = 'mysqlDatabaseName',
     'table-name' = 'mysqlTableName'
   );
   
   create table printSink(
     order_id string,
     order_channel string,
     order_time string,
     pay_amount double,
     real_pay double,
     pay_time string,
     user_id string,
     user_name string,
     area_id STRING,
     primary key(order_id) not enforced
   ) with (
     'connector' = 'print'
   );
   
   insert into printSink select * from mysqlCdcSource;

5. Insert test data in MySQL.

   insert into cdc_order values
   ('202103241000000001','webShop','2021-03-24 10:00:00','100.00','100.00','2021-03-24 10:02:03','0001','Alice','330106'),
   ('202103241606060001','appShop','2021-03-24 16:06:06','200.00','180.00','2021-03-24 16:10:06','0001','Alice','330106');
   
   delete from cdc_order  where order_channel = 'webShop';
   
   insert into cdc_order values('202103251202020001','miniAppShop','2021-03-25 12:02:02','60.00','60.00','2021-03-25 12:03:00','0002','Bob','330110');

6. Perform the following operations to view the data result in the taskmanager.out file:
   1. Log in to the DLI console. In the navigation pane, choose Job Management > Flink Jobs.
   2. Click the name of the corresponding Flink job, choose Run Log, click OBS Bucket, and locate the folder of the log you want to view according to the date.
   3. Go to the folder of the date, find the folder whose name contains taskmanager, download the taskmanager.out file, and view result logs.

   The data result is as follows:

   +I(202103241000000001,webShop,2021-03-2410:00:00,100.0,100.0,2021-03-2410:02:03,0001,Alice,330106)
   +I(202103241606060001,appShop,2021-03-2416:06:06,200.0,180.0,2021-03-2416:10:06,0001,Alice,330106)
   -D(202103241000000001,webShop,2021-03-2410:00:00,100.0,100.0,2021-03-2410:02:03,0001,Alice,330106)
   +I(202103251202020001,miniAppShop,2021-03-2512:02:02,60.0,60.0,2021-03-2512:03:00,0002,Bob,330110)

FAQ

Q: How do I perform window aggregation if the MySQL CDC source table does not support definition of watermarks?

A: You can use the non-window aggregation method. That is, convert the time field into a window value, and then use GROUP BY to perform aggregation based on the window value.

For example, you can use the following script to collect statistics on the number of orders per minute (order_time indicates the order time, in the string format):

insert into printSink select DATE_FORMAT(order_time, 'yyyy-MM-dd HH:mm'), count(*) from mysqlCdcSource group by DATE_FORMAT(order_time, 'yyyy-MM-dd HH:mm');