共计 3241 个字符,预计需要花费 9 分钟才能阅读完成。
本文首发于我的集体博客网站 期待下一个秋 -Flink
什么是 CDC?
CDC 是(Change Data Capture 变更数据获取)的简称。核心思想是,监测并捕捉数据库的变动(包含数据 或 数据表的插入 INSERT、更新 UPDATE、删除 DELETE 等),将这些变更按产生的程序残缺记录下来,写入到消息中间件中以供其余服务进行订阅及生产。
1. 环境筹备
- mysql
- kafka 2.3
-
flink 1.13.5 on yarn
阐明:如果没有装置 hadoop,那么能够不必 yarn,间接用 flink standalone 环境吧。
2. 下载下列依赖包
上面两个地址下载 flink 的依赖包,放在 lib 目录上面。
- flink-sql-connector-kafka_2.11-1.13.5.jar
- flink-sql-connector-mysql-cdc-1.3.0.jar
如果你的 Flink 是其它版本,能够来这里下载。
这里 flink-sql-connector-mysql-cdc,后面一篇文章我用的 mysq-cdc 是 1.4 的,过后是能够的,然而明天我发现须要 mysql-cdc-1.3.0 了,否则,整合 connector-kafka 会有来抵触,目前 mysql-cdc-1.3 适用性更强,都能够兼容的。
如果你是更高版本的 flink,能够自行 https://github.com/ververica/… 下载新版 mvn clean install -DskipTests 本人编译。
这是我编译的最新版 2.2,传上去发现太新了,如果从新换个版本,我得去 gitee 下载源码,不然 github 速度太慢了,而后用 IDEA 编译打包,又得下载一堆依赖。我投降,我间接去网上下载了个 1.3 的间接用了。
我下载的 jar 包,放在 flink 的 lib 目录上面:
flink-sql-connector-kafka_2.11-1.13.5.jar | |
flink-sql-connector-mysql-cdc-1.3.0.jar |
3. 启动 flink-sql client
1) 先在 yarn 下面启动一个 application,进入 flink13.5 目录,执行:
bin/yarn-session.sh -d -s 1 -jm 1024 -tm 2048 -qu root.sparkstreaming -nm flink-cdc-kafka
2) 进入 flink sql 命令行
bin/sql-client.sh embedded -s flink-cdc-kafka
4. 同步数据
这里有一张 mysql 表:
CREATE TABLE `product_view` (`id` int(11) NOT NULL AUTO_INCREMENT, | |
`user_id` int(11) NOT NULL, | |
`product_id` int(11) NOT NULL, | |
`server_id` int(11) NOT NULL, | |
`duration` int(11) NOT NULL, | |
`times` varchar(11) NOT NULL, | |
`time` datetime NOT NULL, | |
PRIMARY KEY (`id`), | |
KEY `time` (`time`), | |
KEY `user_product` (`user_id`,`product_id`) USING BTREE, | |
KEY `times` (`times`) USING BTREE | |
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; | |
-- 样本数据 | |
INSERT INTO `product_view` VALUES ('1', '1', '1', '1', '120', '120', '2020-04-24 13:14:00'); | |
INSERT INTO `product_view` VALUES ('2', '1', '1', '1', '120', '120', '2020-04-24 13:14:00'); | |
INSERT INTO `product_view` VALUES ('3', '1', '1', '3', '120', '120', '2020-04-24 13:14:00'); | |
INSERT INTO `product_view` VALUES ('4', '1', '1', '2', '120', '120', '2020-04-24 13:14:00'); | |
INSERT INTO `product_view` VALUES ('5', '8', '1', '1', '120', '120', '2020-05-14 13:14:00'); | |
INSERT INTO `product_view` VALUES ('6', '8', '1', '2', '120', '120', '2020-05-13 13:14:00'); | |
INSERT INTO `product_view` VALUES ('7', '8', '1', '3', '120', '120', '2020-04-24 13:14:00'); | |
INSERT INTO `product_view` VALUES ('8', '8', '1', '3', '120', '120', '2020-04-23 13:14:00'); | |
INSERT INTO `product_view` VALUES ('9', '8', '1', '2', '120', '120', '2020-05-13 13:14:00'); |
1) 创立数据表关联 mysql
CREATE TABLE product_view_source ( | |
`id` int, | |
`user_id` int, | |
`product_id` int, | |
`server_id` int, | |
`duration` int, | |
`times` string, | |
`time` timestamp, | |
PRIMARY KEY (`id`) NOT ENFORCED | |
) WITH ( | |
'connector' = 'mysql-cdc', | |
'hostname' = '192.168.1.2', | |
'port' = '3306', | |
'username' = 'bigdata', | |
'password' = 'bigdata', | |
'database-name' = 'test', | |
'table-name' = 'product_view' | |
); |
这样,咱们在 flink sql client 操作这个表相当于操作 mysql 外面的对应表。
2) 创立数据表关联 kafka
CREATE TABLE product_view_kafka_sink( | |
`id` int, | |
`user_id` int, | |
`product_id` int, | |
`server_id` int, | |
`duration` int, | |
`times` string, | |
`time` timestamp, | |
PRIMARY KEY (`id`) NOT ENFORCED | |
) WITH ( | |
'connector' = 'upsert-kafka', | |
'topic' = 'flink-cdc-kafka', | |
'properties.bootstrap.servers' = '192.168.1.2:9092', | |
'properties.group.id' = 'flink-cdc-kafka-group', | |
'key.format' = 'json', | |
'value.format' = 'json' | |
); |
这样,kafka 外面的 flink-cdc-kafka 这个主题会被主动创立,如果想指定一些属性,能够提前手动创立好主题,咱们操作表 product_view_kafka_sink,往里面插入数据,能够发现 kafka 中曾经有数据了。
3) 同步数据
建设同步工作,能够应用 sql 如下:
insert into product_view_kafka_sink select * from product_view_source;
这个时候是能够退出 flink sql-client 的,而后进入 flink web-ui,能够看到 mysql 表数据曾经同步到 kafka 中了,对 mysql 进行插入,kafka 都是同步更新的。
通过 kafka 控制台生产,能够看到数据曾经从 mysql 同步到 kafka 了:
参考资料
https://ververica.github.io/f…