flink-cdc同步mysql数据到elasticsearch

阅读原文时间：2023年08月11日阅读：4

CDC是（Change Data Capture 变更数据获取）的简称。核心思想是，监测并捕获数据库的变动（包括数据或数据表的插入INSERT、更新UPDATE、删除DELETE等），将这些变更按发生的顺序完整记录下来，写入到消息中间件中以供其他服务进行订阅及消费。

项目地址：https://github.com/ververica/flink-cdc-connectors

项目文档：https://ververica.github.io/flink-cdc-connectors/master/

mysql
elasticsearch
flink on yarn

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

本例使用版本如下：

下面两个地址下载flink的依赖包，放在lib目录下面。　

　　下载地址：

　　1、https://repo.maven.apache.org/maven2/com/alibaba/ververica/

　　flink-sql-connector-mysql-cdc-1.4.0.jar

　　此仓库提供的最新版本为1.4.0，如需新版本可自行编译或者去https://mvnrepository.com/下载。

　　2、https://repo.maven.apache.org/maven2/org/apache/flink/

　　flink-sql-connector-elasticsearch7_2.11-1.13.5.jar

　　小坑：此处使用的是es7，由于本地环境是es8导致无法创建索引，又重新安装es7测试成功。

启动flink集群

./start-cluster.sh

启动成功的话，可以在 http://localhost:8081/ 访问到 Flink Web UI，如下所示：

启动flink sql client

./sql-client.sh

启动成功后，可以看到如下的页面：

1）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');

2）flink 创建source数据库关联表

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' = '10.34.100.209',
'port' = '3306',
'username' = 'root',
'password' = '123',
'database-name' = 'flinkcdc_test',
'table-name' = 'product_view',
'server-id' = '5401'
);

这样，我们在flink-sql client操作这个表相当于操作mysql里面的对应表。

3）flink 创建sink，数据库关联表elasticsearch

CREATE TABLE product_view_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' = 'elasticsearch-7',
'hosts' = 'http://10.34.100.156:9200',
'index' = 'product_view_index'
);

这样，es里的product_view_index这个索引在数据同步时会被自动创建，如果想指定一些属性，可以提前手动创建好索引。往product_view_sink里面插入数据，可以发现es中已经有数据了。

查看flink创建的表