基于Flink CDC实时同步数据（MySQL到MySQL）

-- 假设我们有两个MySQL实例，source_instance和target_instance
-- 我们将使用Flink CDC来实时同步source_instance中的数据到target_instance
 
-- 首先，在source_instance上为Flink CDC启动器创建用户并授权
CREATE USER 'flinkcdc'@'%' IDENTIFIED BY 'your_password';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'flinkcdc'@'%';
FLUSH PRIVILEGES;
 
-- 然后，在source_instance上为Flink CDC启动器准备binlog
SET GLOBAL binlog_checksum = 'NONE';
SET GLOBAL binlog_format = 'ROW';
SET GLOBAL binlog_row_image = 'FULL';
 
-- 接下来，在Flink上运行以下SQL来实时同步数据
 
CREATE TABLE source_mysql (
    id INT,
    name VARCHAR(255),
    cdc_time TIMESTAMP(3),
    cdc_op_type STRING,
    cdc_before_op_type STRING,
    cdc_update_fields ARRAY<STRING>,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'source_instance_host',
    'port' = '3306',
    'username' = 'flinkcdc',
    'password' = 'your_password',
    'database-name' = 'source_database_name',
    'table-name' = 'source_table_name'
);
 
CREATE TABLE target_mysql (
    id INT,
    name VARCHAR(255),
    cdc_time TIMESTAMP(3),
    cdc_op_type STRING,
    cdc_before_op_type STRING,
    cdc_update_fields ARRAY<STRING>,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://target_instance_host:3306/target_database_name',
    'table-name' = 'target_table_name',
    'username' = 'your_username',
    'password' = 'your_password'
);
 
INSERT INTO target_mysql
SELECT * FROM source_mysql;

这个示例展示了如何使用Flink CDC连接器来同步MySQL数据库中的数据。首先，我们在源数据库上创建了一个用于Flink CDC的用户，并设置了必要的binlog参数。然后，我们定义了源和目标MySQL表，并使用Flink SQL的INSERT INTO ... SELECT ...语句来实现实时数据同步。这个例子非常简单，但它展示了如何将Flink CDC应用于实际场景，并且是学习Flink CDC和实时数据同步技术的一个很好的起点。