Flink-cdc 同步mysql数据

下载地址：https://github.com/ververica/flink-cdc-connectors/releases

这里下载2.2.0版本：https://github.com/ververica/flink-cdc-connectors/archive/refs/tags/release-2.2.0.zip

下载完成后，在 pom.xml 中找到这一项：flink.version ，修改 flink 版本号为：

<flink.version>1.13.6</flink.version>

自行打包编译

通过flink-cdc 同步mysql数据

1、flink集群准备

wget http://mirrors.cloud.tencent.com/apache/flink/flink-1.13.6/flink-1.13.6-bin-scala_2.11.tgz
tar zxvf flink-1.13.6-bin-scala_2.11.tgz

将打包好的 flink-sql-connector-mysql-cdc-2.2-SNAPSHOT.jar 包放入到flink的lib目录下
启动集群

cd flink-1.13.6
bin/start-cluster.sh

2、mysql环境准备

CREATE DATABASE mydb;USE mydb;CREATE TABLE products (id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,name VARCHAR(255) NOT NULL,description VARCHAR(512));ALTER TABLE products AUTO_INCREMENT = 101;INSERT INTO productsVALUES (default,"scooter","Small 2-wheel scooter"),(default,"car battery","12V car battery"),(default,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3"),(default,"hammer","12oz carpenter's hammer"),(default,"hammer","14oz carpenter's hammer"),(default,"hammer","16oz carpenter's hammer"),(default,"rocks","box of assorted rocks"),(default,"jacket","water resistent black wind breaker"),(default,"spare tire","24 inch spare tire");

3、启动flinksql client

cd /opt/flink-1.13.6
bin/sql-client.sh

4、在flinksql client中执行命令

Flink SQL> SET execution.checkpointing.interval = 3s
Flink SQL> CREATE TABLE products (id INT,name STRING,description STRING,PRIMARY KEY (id) NOT ENFORCED) WITH ('connector' = 'mysql-cdc','hostname' = '自己的ip地址','port' = '3306','username' = 'root','password' = '密码','database-name' = 'mydb','table-name' = 'products');Flink SQL> select * from products;

5、在 MySQL 客户端继续插入数据

INSERT INTO products VALUES (default,"scooter1","Small 2-wheel scooter");
INSERT INTO products VALUES (default,"scooter2","Small 2-wheel scooter");
INSERT INTO products VALUES (default,"scooter3","Small 2-wheel scooter");
INSERT INTO products VALUES (default,"scooter4","Small 2-wheel scooter");

4、在flinksql client中查看数据

select * from products;

可以查看到数据变化

Flink-cdc 同步mysql数据相关推荐

Flink CDC入门实践--基于 Flink CDC 构建 MySQL 和 Postgres 的 Streaming ETL
文章目录前言 1.环境准备 2.准备数据 2.1 MySQL 2.2 postgres 3.启动flink和flink sql client 3.1启动flink 3.2启动flink SQL cl ...
技术解析｜Doris Connector 结合 Flink CDC 实现 MySQL 分库分表 Exactly Once精准接入
本篇文档将演示如何使用 Apache Doris Flink Connector 结合 Flink CDC 以及 Doris Stream Load 的两阶段提交,实现 MySQL 数据库分库分表实时 ...
Doris Connector 结合 Flink CDC 实现 MySQL 分库分表 Exactly Once精准接入
导读:本篇文档将演示如何使用 Apache Doris Flink Connector 结合 Flink CDC 以及 Doris Stream Load 的两阶段提交,实现 MySQL 数据库分库分 ...
logstash增量同步mysql数据到es
本篇本章地址:https://www.cnblogs.com/Thehorse/p/11601013.html 今天我们来讲一下logstash同步mysql数据到es 我认为呢,logstash是众 ...
Elasticsearch7.9集群部署，head插件，canal同步mysql数据到es，亲自测试，无坑
Elasticsearch集群部署 1.服务器规划 10.4.7.11 node1 10.4.7.12 node2 10.4.7.13 node3 1. 集群相关一个运行中的 Elastics ...
关于flink cdc 抽取oracle数据 oracle表名大小写的问题
使用flink cdc 抽取oracle数据报错ALTER TABLE XXXXX ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS; 我使用的是flink cdc的 ...
利用Canal全量/增量同步mysql数据至ES
Canal同步mysql数据至ES 1.更改Mysql配置 1.1 开启 Binlog 写入功能配置 binlog-format 为 ROW 模式,配置my.cnf [mysqld] log-bin ...
Logstash同步mysql数据
Logstash同步mysql数据到ElasticSearch 1.安装logstash 下载logstash,解压压缩包下载地址:https://www.elastic.co/downloads/ ...
使用canal实时同步MySQL数据到Elasticsearch
使用canal实时同步MySQL数据到Elasticsearch 搭建环境安装 elasticsearch 安装 kibana 下载和安装canal 1.下载canal 2.配置MySQL 3.配置 ...

Flink-cdc 同步mysql数据

通过flink-cdc 同步mysql数据

Flink-cdc 同步mysql数据相关推荐

最新文章

热门文章