赞
踩
Flink CDC有两种方式同步数据库:
本方案使用FlinkSQL方法,同步两表中的数据。
其中Flink应用可以部署在具有公网IP的服务器上,同时可以连接其他局域网中服务器的数据进行同步工作,如不需要操作管理页面,则不对服务器IP有要求。
将Flink压缩包解压到服务器指定位置
Flink下载地址,点击进入
进入到 根目录\conf\flink-conf.yaml文件,找到rest.port,将其配置为已开放的端口
找到rest.port位置,将其参数localhost改为0.0.0.0,不修改会导致管理页面无法正常打开。
上面的是同步到目标mysql库的连接器
下面的是源mysql的连接器
下载地址:
flink-connector-jdbc:jar:3.0.0-1.16
flink-sql-connector-mysql-cdc:jar:2.3.0
进入到bin文件夹中
1. 运行start-cluster.sh
2. 运行sql-client.sh进入到FlinkSQL模式
创建两个表映射、一个启动任务指令
1)源数据库映射
create table ny_energy_data_source ( id bigint , enterprise_id bigint , use_time timestamp , date_type int , attribute_id bigint , PRIMARY KEY (`id`) NOT ENFORCED )WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'ip地址', 'port' = '3306', 'username' = '用户名', 'password' = '密码', 'database-name' = '源数据库名', 'table-name' = '表名' );
2)目标数据库映射
create table ny_energy_data_target ( id bigint , enterprise_id bigint , use_time timestamp , date_type int , attribute_id bigint , PRIMARY KEY (`id`) NOT ENFORCED )WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://IP地址:3306/库名?serverTimezone=UTC', 'username' = '用户名', 'password' = '密码', 'table-name' = '表名', 'driver' = 'com.mysql.cj.jdbc.Driver', 'scan.fetch-size' = '200' );
3)启动任务
从ny_energy_data_source到ny_energy_data_target,先全量后增量
insert into ny_energy_data_target select * from ny_energy_data_source;
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。