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

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

使用flinkcdc同步mysql至mysql的数据,只会同步一次,修改源表后目标表没有变化
(图片来源网络,侵删)

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") // 根据实际需求设置源表的格式描述符和连接器名称,这里使用默认值即可创建源表的流式表源对象;然后使用类似的方式创建目标表的流式表源对象。

原创文章,作者:未希,如若转载,请注明出处:https://www.kdun.com/ask/580781.html

本网站发布或转载的文章及图片均来自网络,其原创性以及文中表达的观点和判断不代表本网站。如有问题,请联系客服处理。

(0)
未希的头像未希新媒体运营
上一篇 2024-05-07 13:06
下一篇 2024-05-07 13:08

相关推荐

发表回复

您的电子邮箱地址不会被公开。 必填项已用 * 标注

产品购买 QQ咨询 微信咨询 SEO优化
分享本页
返回顶部
云产品限时秒杀。精选云产品高防服务器,20M大带宽限量抢购 >>点击进入