本文首发于我的个人博客网站 等待下一个秋-Flink
什么是CDC？CDC是（Change Data Capture 变更数据获取）的简称。核心思想是，监测并捕获数据库的变动（包括数据 或 数据表的插入INSERT、更新UPDATE、删除DELETE等），将这些变更按发生的顺序完整记录下来，写入到消息中间件中以供其他服务进行订阅及消费。 
1. 环境准备mysql hbase flink 1.13.5 on yarn
说明：如果没有安装hadoop，那么可以不用yarn，直接用flink standalone环境吧。 2. 下载下列依赖包下面两个地址下载flink的依赖包，放在lib目录下面。 - flink-sql-connector-hbase-1.4_2.11-1.13.5.jar
- flink-sql-connector-mysql-cdc-1.4.0.jar
如果你的Flink是其它版本，可以来这里下载。 我是flink1.13，这里flink-sql-connector-mysql-cdc，需要1.4.0以上版本。 
如果你是更高版本的flink，可以自行https://github.com/ververica/flink-cdc-connectors下载新版mvn clean install -DskipTests 自己编译。 
这是我编译的最新版2.2，传上去发现太新了，如果重新换个版本，我得去gitee下载源码，不然github速度太慢了，然后用IDEA编译打包，又得下载一堆依赖。我投降，我直接去网上下载了个1.4的直接用了。 我下载的jar包，放在flink的lib目录下面： 
flink-sql-connector-hbase-1.4_2.11-1.13.5.jar
flink-sql-connector-mysql-cdc-1.4.0.jar
- 先在yarn上面启动一个application，进入flink13.5目录，执行：
bin/yarn-session.sh -d -s 2 -jm 1024 -tm 2048 -qu root.sparkstreaming -nm flink-cdc-hbase
- 进入flink sql命令行
bin/sql-client.sh embedded -s flink-cdc-hbase
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');
- 创建数据表关联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里面的对应表。 - 创建数据表关联hbase
CREATE TABLE product_view_hbase (
rowkey INT,
family1 ROW<user_id INT, product_id INT, server_id INT, duration
关闭
站长推荐 /1 
|
