使用flinkcdc同步mysql至mysql的数据，只会同步一次，修改源表后目标表没有变化

要使用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
);
INSERT INTO source_table (id, name, age) VALUES (1, '张三', 25);
INSERT 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") // 根据实际需求设置源表的格式描述符和连接器名称，这里使用默认值即可创建源表的流式表源对象；然后使用类似的方式创建目标表的流式表源对象。