flink cdc mysql整理与总结

文章目录

  • 一、业务中常见的需要数据同步的场景
    • CDC是什么
    • FlinkCDC是什么
    • CDC原理
    • 为什么是FlinkCDC
    • 业务场景
    • flink cdc对应flink的版本
  • 二、模拟案例
    • 1.阿里云flink sql
    • 2.开源flink sql(单机模式)
    • flink 安装
    • 安装mysql
    • 3.flink datastream
  • 三、总结


提示:以下是本篇文章正文内容,下面案例可供参考

一、业务中常见的需要数据同步的场景

1、多个库的表合并到一张表。不同的业务线或者微服务在不同的数据库里开发,但是此时有些报表需要将多个库的类似的数据合并后做查询统计。或者,某些历史原因,类似刚开始的商业模式不清晰,导致一些业务线分分合合。或者某些边缘业务逐步融合到了主业务。早起的数据是分开的,业务运营也是分开,后来又合并成了一个大块业务。

2、某个数据需要写到多个存储中。业务数据需要写入到多个中间件或者存储中,比如业务的数据存储再Mysql的数据中,后来为了方便检索需要写入到ES,或者为了缓存需要写入到Redis,或者是Mysql分表的数据合并写入到Doris中。

3、数据仓库的场景。比如将表里的数据实时写入到DWS数据仓库的宽表中。

4、应急场景。如果不采专用CDC的方案,那么要达到实时查询的效果,只能在BFF层的代码调用多个中心层的查询API,然后再BFF层做各种聚合,运算。这种方式开发效率低下,万一有的中心层没有提供合适的查询API,临时开发的话,会让开发进度不可控。

总之,不管是数据多写、还是多表合并、还是建立数据仓库,都属于数据同步任务。

示例:pandas 是基于NumPy 的一种工具,该工具是为了解决数据分析任务而创建的。

CDC是什么

CDC 是变更数据捕获(Change Data Capture)技术的缩写,它可以将源数据库(Source)的增量变动记录,同步到一个或多个数据目的(Sink)。在同步过程中,还可以对数据进行一定的处理,例如过滤、关联、分组、统计等。

目前专业做数据库事件接受和解析的中间件是Debezium,如果是捕获Mysql,还有Canal。

FlinkCDC是什么

官网地址:官网FlinkCDC

官方定义:This project provides a set of source connectors for Apache Flink® directly ingesting changes coming from different databases using Change Data Capture(CDC)。根据FlinkCDC官方给出的定义,FlinkCDC提供一组源数据的连接器,使用变更数据捕获的方式,直接吸收来自不同数据库的变更数据。

CDC原理

CDC的原理是,当数据源表发生变动时,会通过附加在表上的触发器或者 binlog 等途径,将操作记录下来。下游可以通过数据库底层的协议,订阅并消费这些事件,然后对数据库变动记录做重放,从而实现同步。这种方式的优点是实时性高,可以精确捕捉上游的各种变动。

为什么是FlinkCDC

1、FlinkCDC 提供了对 Debezium 连接器的封装和集成,简化了配置和使用的过程,并提供了更高级的 API 和功能,例如数据格式转换、事件时间处理等。Flink CDC 使用 Debezium 连接器作为底层的实现,将其与 Flink 的数据处理能力结合起来。通过配置和使用 Flink CDC,您可以轻松地将数据库中的变化数据流转化为 Flink 的 DataStream 或 Table,并进行实时的数据处理、转换和分析。

2、Flink的DataStream和SQL比较成熟和易用

3、Flink支持状态后端(State Backends),允许存储海量的数据状态

4、Flink有更好的生态,更多的Source和Sink的支持

业务场景

  • 数据合并流向:
    在这里插入图片描述
  • 数据多写流向:
    在这里插入图片描述
  • 单数据源写单表流向:
    在这里插入图片描述
  • 数据链路对比
    通过下图,我们可以看到Canal处理数据的链路比FlinkCDC更长,数据链路一旦变长意味着,出错的可能性更高。
    在这里插入图片描述
    在这里插入图片描述

flink cdc对应flink的版本

在这里插入图片描述

二、模拟案例

1.阿里云flink sql

  • 验证mysql开启binlog
    在这里插入图片描述
    flink sql 定义binlog源数据,拿到数据处理(业务逻辑)再写表,后面很简单了
    在这里插入图片描述

2.开源flink sql(单机模式)

背景:win10电脑安装vmware(虚拟化)软件,虚拟机中安装
linux节点一个,flink,mysql(yum默认安装的最新版本:8.0.37),java环境(此处安装java环境略,网上有)

flink 安装

###在linux下载flink包
[root@slave2 ~]# wget https://dlcdn.apache.org/flink/flink-1.18.1/flink-1.18.1-bin-scala_2.12.tgz
### 加压包到当前目录下
[root@slave2 ~]# tar zxvf flink-1.16.3-bin-scala_2.12.tgz 
[root@slave2 ~]# cd flink-1.16.3
[root@slave2 lib]# wget https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mysql-cdc/2.4.0/flink-sql-connector-mysql-cdc-2.4.0.jar修改flink 文件
[root@slave2 flink-1.16.3]# cat conf/flink-conf.yaml 修改内容如下(如果不修改则win10本地电脑无法访问flink web UI,这里浪费很多时间):
taskmanager.host: localhost
rest.bind-address: 0.0.0.0###启动flink
[root@slave2 flink-1.16.3]# bin/start-cluster.sh
###启动客户端
root@slave2 flink-1.16.3]# ./bin/sql-client.sh embedded 

在这里插入图片描述

安装mysql

安装mysql遇到很多小问题

[root@slave2 ~]# yum -y install mysql-community-server
“MySQL 8.0 Community Server” 的 GPG 密钥已安装,但是不适用于此软件包。请检查源的公钥 URL 是否配置正确。

失败的软件包是:mysql-community-libs-8.0.37-1.el7.x86_64

在这里插入图片描述

###用于跳过GPG签名检查 可以安装成功
[root@slave2 ~]# yum -y install mysql-server --nogpgcheck   
###验证myqsl是否可用
[root@slave2 ~]# systemctl start mysqld
[root@slave2 ~]# mysql -u root -p
Enter password: 
Welcome to the MySQL monitor.  Commands end with ; or \g.
Your MySQL connection id is 63
Server version: 8.0.37 MySQL Community Server - GPL
Copyright (c) 2000, 2024, Oracle and/or its affiliates.
Type 'help;' or '\h' for help. Type '\c' to clear the current input statement.
mysql> ### 配置开启binlog
[root@slave2 ~]# vim /etc/my.cnf
log_bin=mysql_bin 
binlog-format=Row 
server-id=1   
###重启mysql
[root@slave2 ~]# systemctl restart mysqld
###重新登陆mysql并查看binlog开启情况
mysql> SHOW VARIABLES LIKE 'log_bin';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| log_bin       | ON    |
+---------------+-------+
1 row in set (0.01 sec)

在这里插入图片描述
mysql创建新用户等

###创建新用户时发现一直报错。查看mysql建密码时要求比较高(新版本mysql对密码要求高)
mysql> SHOW VARIABLES LIKE 'validate_password%';
+-------------------------------------------------+--------+
| Variable_name                                   | Value  |
+-------------------------------------------------+--------+
| validate_password.changed_characters_percentage | 0      |
| validate_password.check_user_name               | ON     |
| validate_password.dictionary_file               |        |
| validate_password.length                        | 10     |
| validate_password.mixed_case_count              | 1      |
| validate_password.number_count                  | 1      |
| validate_password.policy                        | MEDIUM |
| validate_password.special_char_count            | 1      |
+-------------------------------------------------+--------+
8 rows in set (0.00 sec)
###修改密码要求(在工作生产环境不建议这么做)
mysql> SET GLOBAL validate_password.length = 3;
Query OK, 0 rows affected (0.03 sec)
mysql> SET GLOBAL validate_password.policy = LOW;
Query OK, 0 rows affected (0.00 sec)
###再次查看对新建用户密码要求
mysql> SHOW VARIABLES LIKE 'validate_password%';
+-------------------------------------------------+-------+
| Variable_name                                   | Value |
+-------------------------------------------------+-------+
| validate_password.changed_characters_percentage | 0     |
| validate_password.check_user_name               | ON    |
| validate_password.dictionary_file               |       |
| validate_password.length                        | 4     |
| validate_password.mixed_case_count              | 1     |
| validate_password.number_count                  | 1     |
| validate_password.policy                        | LOW   |
| validate_password.special_char_count            | 1     |
+-------------------------------------------------+-------+
8 rows in set (0.00 sec)
###新建用户:root1234
mysql> CREATE USER 'root1234'@'localhost' IDENTIFIED BY 'root1234';
Query OK, 0 rows affected (0.01 sec)mysql> GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'root1234'@'localhost';
Query OK, 0 rows affected (0.02 sec)mysql> GRANT SELECT ON *.* TO 'root1234'@'localhost';
Query OK, 0 rows affected (0.01 sec)mysql> SELECT User, Host FROM mysql.user;
+------------------+-----------+
| User             | Host      |
+------------------+-----------+
| mysql.infoschema | localhost |
| mysql.session    | localhost |
| mysql.sys        | localhost |
| root             | localhost |
| root1234         | localhost |
+------------------+-----------+
5 rows in set (0.00 sec)mysql> FLUSH PRIVILEGES;
Query OK, 0 rows affected (0.00 sec)

建mysql库表(数据源头库表)

Flink SQL>
select * from test_flink_cdc5;
[ERROR] Could not execute SQL statement. Reason:
java.lang.IllegalArgumentException: Can’t find any matched tables, please check your configured database-name: [mysql] and table-name: [mysql.test_cdc]

###mysql 默认只有这4个库(当时直接用默认库mysql建表导致flink 报一些上面奇诡的错)
mysql> show databases;
+--------------------+
| Database           |
+--------------------+
| information_schema |
| mysql              |
| performance_schema |
| sys                |
+--------------------+
###新建一个库:test
mysql> CREATE DATABASE test CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
Query OK, 1 row affected (0.02 sec)mysql> use test;
Database changed
###建表
mysql>  CREATE TABLE `test_cdc` (->   `id` int NOT NULL AUTO_INCREMENT,->   `name` varchar(255) DEFAULT NULL,->   PRIMARY KEY (`id`)->  ) ENGINE=InnoDB ;
Query OK, 0 rows affected (0.04 sec)mysql> show tables;
+----------------+
| Tables_in_test |
+----------------+
| test_cdc       |
+----------------+
1 row in set (0.00 sec)

##开始验证

##mysql 使用上面创建的用户密码登录mysql
[root@slave2 ~]# mysql -u root -p'root1234'
mysql> use test
##flink登录
[root@slave2 flink-1.16.3]# bin/stop-cluster.sh
[root@slave2 flink-1.16.3]# ./bin/sql-client.sh embedded 
##在flink sql中定义mysql源
CREATE TABLE test_flink_cdc ( id INT, name STRING,primary key(id)  NOT ENFORCED
) WITH ('connector' = 'mysql-cdc', 'hostname' = '127.0.0.1', 'port' = '3306', 'username'='root1234', 'password'='root1234', 'database-name'='test', 'table-name'='test_cdc'
);
###查询flink 接收到的binlog数据
Flink SQL> select * from test_flink_cdc;###到mysql sql界面向test_cdc表插入数据
mysql> INSERT INTO test_cdc VALUES (001, 'test01');
Query OK, 1 row affected (0.02 sec)mysql> INSERT INTO test_cdc VALUES (002, 'test02');
Query OK, 1 row affected (0.01 sec)mysql> INSERT INTO test_cdc VALUES (003, 'test03');
Query OK, 1 row affected (0.04 sec)mysql> INSERT INTO test_cdc VALUES (004, 'test04');
Query OK, 1 row affected (0.01 sec)mysql> INSERT INTO test_cdc VALUES (005, 'test05');
Query OK, 1 row affected (0.01 sec)mysql> INSERT INTO test_cdc VALUES (006, 'test06');
Query OK, 1 row affected (0.00 sec)


向mysql表中插入数据
在这里插入图片描述
flink sql这时可以接到binlog数据
在这里插入图片描述
查看flink UI job情况
在这里插入图片描述
在这里插入图片描述
小结:当flink可以拿到mysql binlog源头数据,下面就好做了,根据自己的业务处理sink到任何数据库或组件中(例如sink到mysql,hbase,hive,pg,kafka等等),后面sink就不演示了。

下载链接:
1.mysql jdbc jar包驱动下载
2.flink cdc驱动下载
3.flink下载

3.flink datastream

datastream 比较灵活简单,下面是举例代码片段(datastream的CDC比flink sql还简单,打个jar包在flink web UI界面上传运行即可,此处不做举例)

public class MySqlSourceExample {public static void main(String[] args) throws Exception {MySqlSource<String> mySqlSource = MySqlSource.<String>builder().hostname("yourHostname").port(yourPort).databaseList("yourDatabaseName") // set captured database.tableList("yourDatabaseName.yourTableName") // set captured table.username("yourUsername").password("yourPassword").deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String.build();StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// enable checkpointenv.enableCheckpointing(3000);env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")// set 4 parallel source tasks.setParallelism(4).print().setParallelism(1); // use parallelism 1 for sink to keep message orderingenv.execute("Print MySQL Snapshot + Binlog");}
}

三、总结

  • 公司用的cdc不能分享出来,所以搭建上面案例时遇到很多问题,遇到很多坑。
  • 工作中使用的就是如下截图流程,没有使用canal和kafka,使用的是logtail和SLS(阿里云的组件,类似kafka,但要比kafka等功能强大)
  • 上面说了很多关于flink cdc的优点。我结合工作中使用的flink cdc说一些缺点(自己结合业务场景)。

1.使用flink cdc不适合直观观察binlog的数据(例如脏数据,数据断流,不能直观看到最近的binglog情况,表的更新频率不高等造成的困扰)。


2.使用阿里云SLS收集mysql binlog数据,将数据保存近1个月(保持时间可设置),同时可以在logstore查看数据结构样式(sls支持sql语法查询日志数据),方便后续flink代码开发,也方便flink sql和datastream代码debug。
3.使用flink cdc做数据质量监控比较后知后觉(只能监控已表数据),我这边做法是直接监控sls原始数据质量,发现有数据质量问题会报出来。sls也支持实时断流提醒。
4.一般对于简单的不太重要的业务适合使用flink cdc,这样开发快,数据流不用咋校验。对于复杂数据或复杂业务或重要数据,需要观察binlog数据结果(不信任上游其他部门数据)还是使用类似sls比较好。方便查询数据变化与汇总,方便做数据报警等。
5.总之还是要根据自己的业务场景和自己公司现有技术组件组合着使用比较好。

在这里插入图片描述

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/bicheng/16275.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

kali基本扫描工具(自带)

免责声明:本文仅做技术交流与学习...请勿非法破坏... 详细用法: 命令 -h/百度/翻译 fping 用法 hostlist 文件里面为ip fping -a -q -f hostlist -a 只看存活的 fping -g 202.100.1.1 202.100.1.255 -a -q > Ahost 输出到Ahost文件上 nping nping -c 1 201.100.2.155-244 …

香橙派AIpro使用SSH远程登录

香橙派AIpro可以连接HDMI显示器使用&#xff0c;也可以远程登录。这里采用MobaXterm软件远程登录开发板。 首先要使得控制电脑和香橙派开发板连接到同一个局域网&#xff0c;两者的IP地址能够ping通。在Windows 下可以使用MobaXterm 远程登录开发板&#xff0c;首先新建一个ss…

属于程序员的浪漫,一颗会跳动的心!!!

绘制一颗会跳动的心❤ 嘿嘿 可以说是程序员的专属浪漫了吧&#xff0c;就像点燃一颗LED灯一样&#xff1f;&#xff08;我瞎说的啊&#xff0c;大家别当真&#xff0c;我很菜的&#xff01;&#xff01;&#xff01;&#xff01;&#xff09; 程序就在下面啦&#xff0c;然…

Java8Stream

目录 什么是Stream? IO流&#xff1a; Java8Stream&#xff1a; 什么是流&#xff1f; stream图解 获取流 集合类&#xff0c;使用 Collection 接口下的 stream() 代码 数组类&#xff0c;使用 Arrays 中的 stream() 方法 代码 stream&#xff0c;使用 Stream 中的…

重生之 SpringBoot3 入门保姆级学习(02、打包部署)

重生之 SpringBoot3 入门保姆级学习&#xff08;02、打包部署&#xff09; 1.6 打包插件1.7 测试 jar 包1.8 application.properties 的相关配置 1.6 打包插件 官网链接 https://docs.spring.io/spring-boot/docs/current/reference/html/getting-started.html#getting-starte…

【Python】 XGBoost模型的使用案例及原理解析

原谅把你带走的雨天 在渐渐模糊的窗前 每个人最后都要说再见 原谅被你带走的永远 微笑着容易过一天 也许是我已经 老了一点 那些日子你会不会舍不得 思念就像关不紧的门 空气里有幸福的灰尘 否则为何闭上眼睛的时候 又全都想起了 谁都别说 让我一个人躲一躲 你的承诺 我竟然没怀…

Kubernetes(K8S) 集群环境搭建指南

Kubernetes&#xff08;简称K8s&#xff09;是一个开源的容器编排平台&#xff0c;旨在自动化部署、扩展和管理容器化应用。K8S环境搭建过程比较复杂&#xff0c;涉及到非常多组件安装和系统配置&#xff0c;本文将会详细介绍如何在服务器上搭建好Kubernetes集群环境。 在学习…

面试题·栈和队列的相互实现·详解

A. 用队列实现栈 用队列实现栈 实现代码如下 看着是队列&#xff0c;其实实际实现更接近数组模拟 typedef struct {int* queue1; // 第一个队列int* queue2; // 第二个队列int size; // 栈的大小int front1, rear1, front2, rear2; // 两个队列的首尾指针 } MyS…

6.Redis之String命令

1.String类型基本介绍 redis 所有的 key 都是字符串, value 的类型是存在差异的~~ 一般来说,redis 遇到乱码问题的概率更小~~ Redis 中的字符串,直接就是按照二进制数据的方式存储的. (不会做任何的编码转换【讲 mysql 的时候,知道 mysql 默认的字符集, 是拉丁文,插入中文…

Jenkins--从入门到入土

Jenkins–从入门到入土 文章目录 Jenkins--从入门到入土〇、概念提要--什么是CI/DI&#xff1f;1、CI&#xff08;Continuous Integration&#xff0c;持续集成&#xff09;2、DI&#xff08;DevOps Integration&#xff0c;DevOps 集成&#xff09;3、解决的问题 一、Jenkins安…

world machine学习笔记(4)

选择设备&#xff1a; select acpect&#xff1a; heading&#xff1a;太阳的方向 elevation&#xff1a;太阳的高度 select colour&#xff1a;选择颜色 select convexity&#xff1a;选择突起&#xff08;曲率&#xff09; select height&#xff1a;选择高度 falloff&a…

用常识滚雪球:拼多多的内生价值,九年的变与不变

2024年5月22日&#xff0c;拼多多公布了今年一季度财报&#xff0c;该季度拼多多集团营收868.1亿元&#xff0c;同比增长131%&#xff0c;利润306.0亿&#xff0c;同比增长了202%&#xff0c;数据亮眼。 市场对拼多多经历了“看不见”、“看不懂”、“跟不上”三个阶段。拼多多…

Vue.js条件渲染与列表渲染指南

title: Vue.js条件渲染与列表渲染指南 date: 2024/5/26 20:11:49 updated: 2024/5/26 20:11:49 categories: 前端开发 tags: VueJS前端开发数据绑定列表渲染状态管理路由配置性能优化 第1章&#xff1a;Vue.js基础与环境设置 1.1 Vue.js简介 Vue.js (读音&#xff1a;/vju…

SwiftUI中的Slider的基本使用

在SwiftUI中&#xff0c;可以使用Slider视图创建一个滑动条&#xff0c;允许用户从范围中选择一个值。通过系统提供的Slider&#xff0c;用起来也很方便。 Slider 先看一个最简单的初始化方法&#xff1a; State private var sliderValue: Float 100var body: some View {V…

ollama 使用,以及指定模型下载地址

ollama windows 使用 官网&#xff1a; https://ollama.com/ windows 指定 models 下载地址 默认会下载在C盘 &#xff0c;占用空间 在Windows系统中&#xff0c;可以通过设置环境变量OLLAMA_MODELS来指定模型文件的下载和存储路径。具体操作步骤如下&#xff1a; 1.打开系统…

【九十四】【算法分析与设计】练习四蛮力法练习,排列问题和组合问题,求解最大连续子序列和问题,求解幂集问题,求解0/1背包问题,求解任务分配问题

求解最大连续子序列和问题 给定一个有n&#xff08;n≥1&#xff09;个整数的序列&#xff0c;要求求出其中最大连续子序列的和。 例如&#xff1a; 序列&#xff08;-2&#xff0c;11&#xff0c;-4&#xff0c;13&#xff0c;-5&#xff0c;-2&#xff09;的最大子序列和为20…

第 33 次CCF认证

1. 词频统计 题目描述 样例输入 代码 #include <bits/stdc.h>using namespace std;int main() {int n,m;cin>>n>>m;vector<int> ans1(m,0),ans2(m,0);while (n --) {int t;cin>>t;vector<int> vis(m1,0);for (int i 1;i < t;i ) {i…

数据结构(五)

数据结构&#xff08;五&#xff09; 常见的排序算法内部排序交换插入选择归并基数 外部排序基于归并的 常见的排序算法 内部排序 交换 冒泡&#xff1a;每一次运行总会将最小的或者最大的放到前面&#xff0c;如果需要交换&#xff0c;一直在交换 快速排序*&#xff1a;经过…

2024最新前端面试八股文【基础篇293题】

⼀、HTML、HTTP、web综合问题 1 前端需要注意哪些SEO 2 <img> 的 title 和 alt 有什么区别 3 HTTP的⼏种请求⽅法⽤途 4 从浏览器地址栏输⼊url到显示⻚⾯的步骤 5 如何进⾏⽹站性能优化 6 HTTP状态码及其含义 7 语义化的理解 8 介绍⼀下你对浏览器内核的理解 9 …

【操作系统】发展与分类(手工操作、批处理、分时操作、实时操作)

2.操作系统发展与分类 思维导图 手工操作阶段&#xff08;此阶段无操作系统&#xff09; 需要人工干预 缺点&#xff1a; 1.用户独占全机&#xff0c;资源利用率低&#xff1b; 2.CPU等待手工操作&#xff0c;CPU利用不充分。 批处理阶段&#xff08;操作系统开始出现&#x…