教你如何flink-cdc同步mysql数据到kafka

database 投稿 1600 0 评论

什么是CDC?

CDC 是change data capture,即变化数据捕捉。是数据库进行备份的一种方式,常用于大量数据的备份工作。分为入侵式的和非入侵式的备份方法,入侵式的有基于触发器备份、基于时间戳备份、基于快照备份,非入侵式的备份方法是基于日志的备份。mysql 基于日志的CDC就是要开启mysql binary log。

教你如何flink-cdc同步mysql数据到kafka

1. 环境准备

  • kafka 2.3

说明:如果没有安装hadoop,那么可以不用yarn,直接用flink standalone环境吧。

2. 下载下列依赖包

  1. flink-sql-connector-kafka_2.11-1.13.5.jar

  2. flink-sql-connector-mysql-cdc-1.3.0.jar

如果你的Flink是其它版本,可以来这里下载。

教你如何flink-cdc同步mysql数据到kafka

教你如何flink-cdc同步mysql数据到kafka

我下载的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
  1. 进入flink sql命令行

bin/sql-client.sh embedded -s flink-cdc-kafka

教你如何flink-cdc同步mysql数据到kafka

4. 同步数据

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里面的对应表。

  1. 创建数据表关联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中已经有数据了。

  1. 同步数据

insert into product_view_kafka_sink select * from product_view_source;

这个时候是可以退出flink sql-client的,然后进入flink web-ui,可以看到mysql表数据已经同步到kafka中了,对mysql进行插入,kafka都是同步更新的。

教你如何flink-cdc同步mysql数据到kafka

https://ververica.github.io/flink-cdc-connectors/master/content/about.html

编程学习分享 » 教你如何flink-cdc同步mysql数据到kafka

赞 (0) or 分享 (0)
游客 发表我的评论   换个身份
取消评论

表情
(0)个小伙伴在吐槽

高效,专业,符合SEO

联系我们