要使用Flink CDC同步MySQL至MySQL的数据，首先需要确保已经安装了Flink和Debezium MySQL Connector，接下来，按照以下步骤进行操作：

（图片来源网络，侵删）

1、创建源MySQL数据库的表并插入数据

在源MySQL数据库中创建一个表，并插入一些数据，创建一个名为source_db的数据库，并在其中创建一个名为source_table的表：

CREATE DATABASE source_db;USE source_db;CREATE TABLE source_table (  id INT PRIMARY KEY,  name VARCHAR(255),  age INT);ｉｎｓｅｒｔ INTO source_table (id, name, age) VALUES (1, '张三', 25);ｉｎｓｅｒｔ INTO source_table (id, name, age) VALUES (2, '李四', 30);

2、创建目标MySQL数据库的表

在目标MySQL数据库中创建一个与源表结构相同的表，创建一个名为target_db的数据库，并在其中创建一个名为target_table的表：

CREATE DATABASE target_db;USE target_db;CREATE TABLE target_table (  id INT PRIMARY KEY,  name VARCHAR(255),  age INT);

3、配置Flink CDC源连接器和目标连接器

在Flink应用程序中，配置CDC源连接器和目标连接器，这里以Flink SQL为例：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;import org.apache.flink.table.api.EnvironmentSettings;import org.apache.flink.table.catalog.debezium.DebeziumMySqlCatalog;import org.apache.flink.table.catalog.debezium.DebeziumMySqlOptions;import org.apache.flink.table.descriptors.*;import org.apache.flink.table.sources.StreamTableSource;import org.apache.flink.types.Row;public class FlinkCDCExample {    public static void main(String[] args) throws Exception {        // 创建流处理执行环境        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);        // 注册Debezium MySql源连接器和目标连接器        String sourceDcUrl = "jdbc:mysql://localhost:3306/source_db";        String targetDcUrl = "jdbc:mysql://localhost:3306/target_db";        String user = "root";        String password = "password";        DebeziumMySqlCatalog sourceCatalog = new DebeziumMySqlCatalog(sourceDcUrl, user, password, new DebeziumMySqlOptions());        DebeziumMySqlCatalog targetCatalog = new DebeziumMySqlCatalog(targetDcUrl, user, password, new DebeziumMySqlOptions());        tableEnv.registerCatalog("source", sourceCatalog);        tableEnv.registerCatalog("target", targetCatalog);        tableEnv.useCatalog("source");        tableEnv.useCatalog("target");        // 创建源表和目标表的字段信息和类型信息        FieldSchema sourceIdField = new FieldSchema("id", TypeInformation.of(Types.INT));        FieldSchema sourceNameField = new FieldSchema("name", TypeInformation.of(Types.STRING));        FieldSchema sourceAgeField = new FieldSchema("age", TypeInformation.of(Types.INT));        FieldSchema targetIdField = new FieldSchema("id", TypeInformation.of(Types.INT));        FieldSchema targetNameField = new FieldSchema("name", TypeInformation.of(Types.STRING));        FieldSchema targetAgeField = new FieldSchema("age", TypeInformation.of(Types.INT));        List<FieldSchema> sourceFields = Arrays.asList(sourceIdField, sourceNameField, sourceAgeField);        List<FieldSchema> targetFields = Arrays.asList(targetIdField, targetNameField, targetAgeField);        DataType sourceDataType = DataTypes.createStructType(sourceFields);        DataType targetDataType = DataTypes.createStructType(targetFields);        SingleRowDataSourceFactory dataSourceFactory = new SingleRowDataSourceFactory();        dataSourceFactory.setDataSource(new FlinkKafkaConsumer<>(...)); // 根据实际需求设置Kafka消费者参数        SingleRowTableFactory tableFactory = new SingleRowTableFactory();        tableFactory.setDataSourceFactory(dataSourceFactory);        tableFactory.setRowTypeInfo(sourceDataType);        tableFactory.setKeyFields("id"); // 根据实际需求设置主键字段名        tableFactory.setProducedPartitionsPath("/path/to/produced/partitions"); // 根据实际需求设置生产分区路径        tableFactory.setRetainedFilesTime(Duration.ofDays(7)); // 根据实际需求设置保留文件的时间间隔，默认为7天        tableFactory.setCompactInterval(Duration.ofHours(1)); // 根据实际需求设置压缩间隔，默认为1小时        tableFactory.setMaxCompactedFileSize(1024 * 1024 * 1024L); // 根据实际需求设置最大压缩文件大小，默认为1GB        tableFactory.setDeletedFilesPeriod(Duration.ofDays(7)); // 根据实际需求设置删除文件的周期，默认为7天        tableFactory.setDeletedFilesDirectory("/path/to/deleted/files"); // 根据实际需求设置删除文件的目录路径，默认为"$FLINK_HOME/tmp" + "deletedfiles"目录        StreamTableSource<?> sourceTableSource = tableEnv.from("source_db." + "source_table").withFormatDescription("Debezium MySql Source") // 根据实际需求设置源表的格式描述符和连接器名称，这里使用默认值即可创建源表的流式表源对象；然后使用类似的方式创建目标表的流式表源对象。