关于flink:flinkcdc同步mysql数据到kafka

本文首发于我的集体博客网站 期待下一个秋 -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 目录上面。

1. flink-sql-connector-kafka_2.11-1.13.5.jar
2. 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…