等待下一个秋

  • Spark
  • Flink
  • Hive
  • 数据仓库
  • ClickHouse
  • 收徒弟
  • Java
    • Spring
    • Mybatis
    • SpringBoot
    • 面试题
  • Python
    • Python基础
    • 爬虫
    • Numpy
    • matplotlib
    • Flask
  • 技术杂谈
    • Linux知识
    • Docker
    • Git教程
    • Redis教程
    • mysql
    • 前端
    • R语言
    • 机器学习
  • 关于我
  • 其它
    • 副业挣钱
    • 资料下载
    • 资料文档
专注于Hadoop/Spark/Flink/Hive/数据仓库等
关注公众号:大数据技术派,获取更多学习资料。
  1. 首页
  2. Flink
  3. 正文

flink-cdc同步mysql数据到hbase

2022年9月15日 1399点热度 0人点赞 0条评论

本文首发于我的个人博客网站 等待下一个秋-Flink

什么是CDC?

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

Flink_CDC

1. 环境准备

  • mysql

  • hbase

  • flink 1.13.5 on yarn

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

2. 下载下列依赖包

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

  1. flink-sql-connector-hbase-1.4_2.11-1.13.5.jar
  2. flink-sql-connector-mysql-cdc-1.4.0.jar

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

我是flink1.13,这里flink-sql-connector-mysql-cdc,需要1.4.0以上版本。

image-20220913170030754

如果你是更高版本的flink,可以自行https://github.com/ververica/flink-cdc-connectors下载新版mvn clean install -DskipTests 自己编译。

img

这是我编译的最新版2.2,传上去发现太新了,如果重新换个版本,我得去gitee下载源码,不然github速度太慢了,然后用IDEA编译打包,又得下载一堆依赖。我投降,我直接去网上下载了个1.4的直接用了。

我下载的jar包,放在flink的lib目录下面:

image-20220915145142496

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

1) 先在yarn上面启动一个application,进入flink13.5目录,执行:

bin/yarn-session.sh -d -s 2 -jm 1024 -tm 2048 -qu root.sparkstreaming -nm flink-cdc-hbase

2) 进入flink sql命令行

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

img

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) 创建数据表关联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这个主题。

3) 同步数据

flink-cdc-mysql2hbase

建立同步任务,可以使用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                                                                                                   
 \x00\x00\x00\x01                      column=family1:duration, timestamp=1663223736391, value=\x00\x00\x00x                                         
 \x00\x00\x00\x01                      column=family1:product_id, timestamp=1663223736391, value=\x00\x00\x00\x01                                    
 \x00\x00\x00\x01                      column=family1:server_id, timestamp=1663223736391, value=\x00\x00\x00\x01                                     
 \x00\x00\x00\x01                      column=family1:user_id, timestamp=1663223736391, value=\x00\x00\x00\x01                                       
 \x00\x00\x00\x02                      column=family1:duration, timestamp=1663223736391, value=\x00\x00\x00x                                         
 \x00\x00\x00\x02                      column=family1:product_id, timestamp=1663223736391, value=\x00\x00\x00\x01                                    
 \x00\x00\x00\x02                      column=family1:server_id, timestamp=1663223736391, value=\x00\x00\x00\x01                                     
 \x00\x00\x00\x02                      column=family1:user_id, timestamp=1663223736391, value=\x00\x00\x00\x01                                       
 \x00\x00\x00\x03                      column=family1:duration, timestamp=1663223736391, value=\x00\x00\x00x                                         
 \x00\x00\x00\x03                      column=family1:product_id, timestamp=1663223736391, value=\x00\x00\x00\x01                                    
 \x00\x00\x00\x03                      column=family1:server_id, timestamp=1663223736391, value=\x00\x00\x00\x03                                     
 \x00\x00\x00\x03                      column=family1:user_id, timestamp=1663223736391, value=\x00\x00\x00\x01                                       
 \x00\x00\x00\x04                      column=family1:duration, timestamp=1663223736391, value=\x00\x00\x00x                                         
 \x00\x00\x00\x04                      column=family1:product_id, timestamp=1663223736391, value=\x00\x00\x00\x01                                    
 \x00\x00\x00\x04                      column=family1:server_id, timestamp=1663223736391, value=\x00\x00\x00\x02                                     
 \x00\x00\x00\x04                      column=family1:user_id, timestamp=1663223736391, value=\x00\x00\x00\x01                                       
 \x00\x00\x00\x05                      column=family1:duration, timestamp=1663223736391, value=\x00\x00\x00x                                         
 \x00\x00\x00\x05                      column=family1:product_id, timestamp=1663223736391, value=\x00\x00\x00\x01                                    
 \x00\x00\x00\x05                      column=family1:server_id, timestamp=1663223736391, value=\x00\x00\x00\x01                                     
 \x00\x00\x00\x05                      column=family1:user_id, timestamp=1663223736391, value=\x00\x00\x00\x08                                       
 \x00\x00\x00\x06                      column=family1:duration, timestamp=1663223736391, value=\x00\x00\x00x                                         
 \x00\x00\x00\x06                      column=family1:product_id, timestamp=1663223736391, value=\x00\x00\x00\x01                                    
 \x00\x00\x00\x06                      column=family1:server_id, timestamp=1663223736391, value=\x00\x00\x00\x02                                     
 \x00\x00\x00\x06                      column=family1:user_id, timestamp=1663223736391, value=\x00\x00\x00\x08                                       
 \x00\x00\x00\x07                      column=family1:duration, timestamp=1663223736391, value=\x00\x00\x00x                                         
 \x00\x00\x00\x07                      column=family1:product_id, timestamp=1663223736391, value=\x00\x00\x00\x01                                    
 \x00\x00\x00\x07                      column=family1:server_id, timestamp=1663223736391, value=\x00\x00\x00\x03                                     
 \x00\x00\x00\x07                      column=family1:user_id, timestamp=1663223736391, value=\x00\x00\x00\x08                                       
 \x00\x00\x00\x09                      column=family1:duration, timestamp=1663223736391, value=\x00\x00\x00x                                         
 \x00\x00\x00\x09                      column=family1:product_id, timestamp=1663223736391, value=\x00\x00\x00\x01                                    
 \x00\x00\x00\x09                      column=family1:server_id, timestamp=1663223736391, value=\x00\x00\x00\x03                                     
 \x00\x00\x00\x09                      column=family1:user_id, timestamp=1663223736391, value=\x00\x00\x00\x08                                       
 \x00\x00\x00\x0A                      column=family1:duration, timestamp=1663223736391, value=\x00\x00\x00x                                         
 \x00\x00\x00\x0A                      column=family1:product_id, timestamp=1663223736391, value=\x00\x00\x00\x01                                    
 \x00\x00\x00\x0A                      column=family1:server_id, timestamp=1663223736391, value=\x00\x00\x00\x02                                     
 \x00\x00\x00\x0A                      column=family1:user_id, timestamp=1663223736391, value=\x00\x00\x00\x08                                       
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里面的数据了:

image-20220915153846428

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。

image-20220915160523185

看看查出来的结果吧,这是flnk-sql client:

image-20220915160617440

比如我选中这一行,进来后是这条数据的详细情况,是没有问题的:

image-20220915160731799

参考资料

https://nightlies.apache.org/flink/flink-docs-release-1.13/zh/docs/connectors/table/hbase/

标签: Flink
最后更新:2022年9月15日

等待下一个秋

待我代码写成,便娶你为妻!专注于Hadoop/Spark/Flink/Hive/数据仓库等,关注公众号:大数据技术派,获取更多学习资料。

打赏 点赞
< 上一篇
下一篇 >

文章评论

取消回复

等待下一个秋

待我代码写成,便娶你为妻!专注于Hadoop/Spark/Flink/Hive/数据仓库等,关注公众号:大数据技术派,获取更多学习资料。

搜一搜
微信
最新 热点 随机
最新 热点 随机
ClickHouse 自定义分区键 ClickHouse数据副本引擎 ClickHouse ReplacingMergeTree引擎 ClickHouse MergeTree引擎 clickhouse简介 Flink SQL管理平台flink-streaming-platform-web安装搭建
生成当前Python环境的依赖文件 关于副业,我的一次亏钱经历 第01讲:Flink 的应用场景和架构模型 个人IP、人设 短视频PC录制——OBS studio开源免费的录屏工具 如何使用MySQL Workbench将样本数据库导入到MySQL数据库服务器
标签聚合
算法 Flink 挣钱 Java 大数据 书籍 Redis 数据仓库 mysql Python Hive R语言
文章归档
  • 2022年12月
  • 2022年11月
  • 2022年9月
  • 2022年7月
  • 2022年6月
  • 2022年5月
  • 2022年4月
  • 2022年3月
  • 2022年2月
  • 2022年1月
  • 2021年12月
  • 2021年11月
  • 2021年10月
  • 2021年9月
  • 2021年8月
  • 2021年6月
  • 2021年5月
  • 2021年4月
  • 2021年3月
  • 2021年2月
  • 2021年1月
  • 2020年12月
  • 2020年11月
  • 2020年10月
  • 2020年9月
  • 2020年8月
  • 2020年7月
  • 2020年5月
  • 2020年4月
  • 2020年1月
  • 2019年9月
  • 2019年8月
  • 2019年7月
  • 2019年6月
  • 2019年5月
  • 2019年4月
  • 2019年3月
  • 2019年1月
  • 2018年12月
  • 2017年5月

©2022 ikeguang.com. 保留所有权利。

鄂ICP备2020019097号-1

鄂公网安备 42032202000160号