AnalyticDB:Use Flink to subscribe to binary logs

(Optional) Set the binary log retention period

The default retention period is 6 hours. To change it, use the binlog_ttl parameter:

ALTER TABLE source_table binlog_ttl='1d';

binlog_ttl supports the following formats:

To check the current retention period, run:

SHOW CREATE TABLE source_table;

Create a Flink SQL job

Log on to the Realtime Compute for Apache Flink console and create an SQL job.

Create a source table

Define a source table in Flink that connects to AnalyticDB for MySQL and reads binary log data from source_table:

CREATE TEMPORARY TABLE adb_source (
  `id` INT,
  `num` BIGINT,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'adb-mysql-cdc',
  'hostname' = 'amv-2zepb9n1l58ct01z50000****.ads.aliyuncs.com',
  'username' = 'testUser',
  'password' = 'Test12****',
  'database-name' = 'binlog',
  'table-name' = 'source_table'
);

The primary key defined in the Flink DDL statement must be identical to the primary key in the AnalyticDB for MySQL table, including the key name. Mismatched primary keys affect data correctness.

Create a destination table

Create the physical table in AnalyticDB for MySQL where the processed data will be stored:

CREATE TABLE target_table (
  `id` INT,
  `num` BIGINT,
  PRIMARY KEY (`id`)
)

Create a sink table

Define a Flink sink table that writes processed data to the destination table in AnalyticDB for MySQL:

CREATE TEMPORARY TABLE adb_sink (
  `id` INT,
  `num` BIGINT,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'adb3.0',
  'url' = 'jdbc:mysql://amv-2zepb9n1l58ct01z50000****.ads.aliyuncs.com:3306/flinktest',
  'userName' = 'testUser',
  'password' = 'Test12****',
  'tableName' = 'target_table'
);

For sink table WITH parameters and type mappings, see AnalyticDB for MySQL V3.0 connector.

Sync changes from source to sink

Write a query that reads from the source table and writes to the sink table:

INSERT INTO adb_sink
SELECT * FROM adb_source;

Validate and deploy the job

1. Click Save.

2. Click Validate. Validation checks the job's SQL semantics, network connectivity, and table metadata. In the results area, click SQL Advice to view risk alerts and optimization suggestions.

3. (Optional) Click Debug to simulate the job, check output results, and verify the business logic of SELECT or INSERT statements. For details, see Debug a deployment.

4. Click Deploy. After deployment, go to the O&M page to start the job. For details on deploying, see Create a deployment.