flink-cdc同步mysql数据到hbase

技术分享 4年前 (2022-09-16) 0 999+

本文首发于我的个人博客网站等待下一个秋-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是其它版本，可以来这里下载。

我是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

3. 启动flink-sql client

先在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 INT>,  PRIMARY KEY (rowkey) NOT ENFORCED ) WITH (  'connector' = 'hbase-1.4',  'table-name' = 'product_view',  'zookeeper.quorum' = 'cdh-001:2181' );

这里，需要提前在hbase里面创建好product_view这个主题。

同步数据

建立同步任务，可以使用sql如下：

insert into product_view_hbase select id as rowkey, ROW(user_id, product_id, server_id, duration) from product_view_source;

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

进入hbase shell，可以看到数据已经从mysql同步到hbase了：

hbase(main):009:0> scan 'product_view' ROW                                    COLUMN+CELL                                                                                                     x00x00x00x01                      column=family1:duration, timestamp=1663223736391, value=x00x00x00x                                           x00x00x00x01                      column=family1:product_id, timestamp=1663223736391, value=x00x00x00x01                                      x00x00x00x01                      column=family1:server_id, timestamp=1663223736391, value=x00x00x00x01                                       x00x00x00x01                      column=family1:user_id, timestamp=1663223736391, value=x00x00x00x01                                         x00x00x00x02                      column=family1:duration, timestamp=1663223736391, value=x00x00x00x                                           x00x00x00x02                      column=family1:product_id, timestamp=1663223736391, value=x00x00x00x01                                      x00x00x00x02                      column=family1:server_id, timestamp=1663223736391, value=x00x00x00x01                                       x00x00x00x02                      column=family1:user_id, timestamp=1663223736391, value=x00x00x00x01                                         x00x00x00x03                      column=family1:duration, timestamp=1663223736391, value=x00x00x00x                                           x00x00x00x03                      column=family1:product_id, timestamp=1663223736391, value=x00x00x00x01                                      x00x00x00x03                      column=family1:server_id, timestamp=1663223736391, value=x00x00x00x03                                       x00x00x00x03                      column=family1:user_id, timestamp=1663223736391, value=x00x00x00x01                                         x00x00x00x04                      column=family1:duration, timestamp=1663223736391, value=x00x00x00x                                           x00x00x00x04                      column=family1:product_id, timestamp=1663223736391, value=x00x00x00x01                                      x00x00x00x04                      column=family1:server_id, timestamp=1663223736391, value=x00x00x00x02                                       x00x00x00x04                      column=family1:user_id, timestamp=1663223736391, value=x00x00x00x01                                         x00x00x00x05                      column=family1:duration, timestamp=1663223736391, value=x00x00x00x                                           x00x00x00x05                      column=family1:product_id, timestamp=1663223736391, value=x00x00x00x01                                      x00x00x00x05                      column=family1:server_id, timestamp=1663223736391, value=x00x00x00x01                                       x00x00x00x05                      column=family1:user_id, timestamp=1663223736391, value=x00x00x00x08                                         x00x00x00x06                      column=family1:duration, timestamp=1663223736391, value=x00x00x00x                                           x00x00x00x06                      column=family1:product_id, timestamp=1663223736391, value=x00x00x00x01                                      x00x00x00x06                      column=family1:server_id, timestamp=1663223736391, value=x00x00x00x02                                       x00x00x00x06                      column=family1:user_id, timestamp=1663223736391, value=x00x00x00x08                                         x00x00x00x07                      column=family1:duration, timestamp=1663223736391, value=x00x00x00x                                           x00x00x00x07                      column=family1:product_id, timestamp=1663223736391, value=x00x00x00x01                                      x00x00x00x07                      column=family1:server_id, timestamp=1663223736391, value=x00x00x00x03                                       x00x00x00x07                      column=family1:user_id, timestamp=1663223736391, value=x00x00x00x08                                         x00x00x00x09                      column=family1:duration, timestamp=1663223736391, value=x00x00x00x                                           x00x00x00x09                      column=family1:product_id, timestamp=1663223736391, value=x00x00x00x01                                      x00x00x00x09                      column=family1:server_id, timestamp=1663223736391, value=x00x00x00x03                                       x00x00x00x09                      column=family1:user_id, timestamp=1663223736391, value=x00x00x00x08                                         x00x00x00x0A                      column=family1:duration, timestamp=1663223736391, value=x00x00x00x                                           x00x00x00x0A                      column=family1:product_id, timestamp=1663223736391, value=x00x00x00x01                                      x00x00x00x0A                      column=family1:server_id, timestamp=1663223736391, value=x00x00x00x02                                       x00x00x00x0A                      column=family1:user_id, timestamp=1663223736391, value=x00x00x00x08                                        9 row(s) Took 0.1656 seconds

直接在flink-sql client里面查询hbase数据，也是可以的：

Flink SQL> select * from product_view_hbase ; 2022-09-15 15:38:23,205 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar 2022-09-15 15:38:23,207 INFO  org.apache.hadoop.yarn.client.ConfiguredRMFailoverProxyProvider [] - Failing over to rm72 2022-09-15 15:38:23,212 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - Found Web Interface cdh-001:35225 of application 'application_1633924491541_7321'.

执行上面查询sql，就会进入界面，这就是hbase里面的数据了：

5. 关联查询

在这个flink-sql client环境中，这里有两张表：product_view_source（mysql的表）和product_view_hbase（hbase的表），后者是有前者查询导入的，这里为了简单，我没有再关联其它第三张表，就用这两张表，做关联查询，达到演示的目的。

select product_view_source.*, product_view_hbase.*  from product_view_source inner join product_view_hbase on product_view_source.id = product_view_hbase.rowkey;

这里做了个简单的关联查询，通过id跟rowkey关联，然后打开web-ui，通过flink web-ui结果可以看出，这里是个hash join，两个source，到join，一共3个task。