flinkcdc 3.0 尝鲜

本文会将从环境搭建到demo来全流程体验flinkcdc 3.0
包含了如下内容

  1. flink1.18 standalone搭建
  2. doris 1fe1be 搭建
  3. 整库数据同步
  4. 测试各同步场景
  5. 从检查点重启同步任务

环境搭建

flink环境(Standalone模式)

下载flink 1.18.0 链接 : https://archive.apache.org/dist/flink/flink-1.18.0/flink-1.18.0-bin-scala_2.12.tgz

解压 :

tar -zxvf flink-1.18.0-bin-scala_2.12.tgz

修改checkpoint 时间间隔 为3秒

vim conf/flink-conf.yaml 
# 94 行(set nu 显示行)
taskmanager.numberOfTaskSlots: 2
# 148 行
execution.checkpointing.interval: 3000

启动

./bin/start-cluster.sh

访问页面 : http://127.0.0.1:8081
image.png

doris环境(1fe1be)

修改环境宿主机的内存映射

# 因为mac内部实现容器的方式不同,直接修改max_map_count值可能无法成功,所以在容器中进行修改
docker run -it --privileged --pid=host --name=change_count debian nsenter -t 1 -m -u -n -i sh
# 修改内存映射值(这个值通常用于限制一个进程打开的文件数量,默认是65530)
sysctl -w vm.max_map_count=2000000
# 退出容器
exit

使用docker compose 搭建doris 1fe1be集群

version: '3'
services:docker-fe-01:image: "apache/doris:1.2.2-fe-arm"container_name: "doris-fe-01"hostname: "fe-01"environment:- FE_SERVERS=fe1:172.20.80.2:9010- FE_ID=1ports:- 8031:8030- 9031:9030volumes:- /Users/antg/docker/doris_1fe_1be/data/fe-01/doris-meta:/opt/apache-doris/fe/doris-meta- /Users/antg/docker/doris_1fe_1be/data/fe-01/conf:/opt/apache-doris/fe/conf- /Users/antg/docker/doris_1fe_1be/data/fe-01/log:/opt/apache-doris/fe/lognetworks:doris_net:ipv4_address: 172.20.80.2docker-be-01:image: "apache/doris:1.2.2-be-arm"container_name: "doris-be-01"hostname: "be-01"depends_on:- docker-fe-01environment:- FE_SERVERS=fe1:172.20.80.2:9010- BE_ADDR=172.20.80.5:9050ports:- 8041:8040volumes:- /Users/antg/docker/doris_1fe_1be/data/be-01/storage:/opt/apache-doris/be/storage- /Users/antg/docker/doris_1fe_1be/data/be-01/conf:/opt/apache-doris/be/conf- /Users/antg/docker/doris_1fe_1be/data/be-01/script:/docker-entrypoint-initdb.d- /Users/antg/docker/doris_1fe_1be/data/be-01/log:/opt/apache-doris/be/lognetworks:doris_net:ipv4_address: 172.20.80.5
networks:doris_net:ipam:config:- subnet: 172.20.80.0/24

启动并验证是否启动成功

# 启动
docker-compose -f 1fe_1be.yaml up -d
# 连接doris
mysql -h127.0.0.1 -P9031 -uroot -p
# 创建数据库 doris_sync
> create database doris_sync;

mysql环境及测试数据准备

使用本机之前安装的mysql

建测试库测试表

create database doris_sync;
CREATE TABLE `a_0` (`id` int NOT NULL AUTO_INCREMENT,`name` varchar(255) DEFAULT NULL,PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;CREATE TABLE `a_1` (`id` int NOT NULL AUTO_INCREMENT,`name` varchar(255) DEFAULT NULL,PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;CREATE TABLE `abc` (`id` int NOT NULL AUTO_INCREMENT,`name` varchar(255) DEFAULT NULL,PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=11 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;CREATE TABLE `table_0` (`id` int NOT NULL AUTO_INCREMENT,`name` varchar(255) DEFAULT NULL,PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=11 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;CREATE TABLE `table_1` (`id` int NOT NULL AUTO_INCREMENT,`name` varchar(255) DEFAULT NULL,PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=101 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

其中 a_0,a_1 是分表,table_0,table_1是另外一个分表,abc是一个单独的表

初始化插入一些测试数据

INSERT INTO `a_0` (`id`, `name`) VALUES (1, 'a');
INSERT INTO `a_1` (`id`, `name`) VALUES (2, 'b');
BEGIN;
INSERT INTO `abc` (`id`, `name`) VALUES (1, 'Luo Rui');
INSERT INTO `abc` (`id`, `name`) VALUES (2, 'Yung Wing Kuen');
INSERT INTO `abc` (`id`, `name`) VALUES (3, 'Chiang Chun Yu');
INSERT INTO `abc` (`id`, `name`) VALUES (4, 'Tang Ming');
INSERT INTO `abc` (`id`, `name`) VALUES (5, 'Man Wai Lam');
INSERT INTO `abc` (`id`, `name`) VALUES (6, 'Tin Tsz Ching');
INSERT INTO `abc` (`id`, `name`) VALUES (7, 'Doris Moore');
INSERT INTO `abc` (`id`, `name`) VALUES (8, 'Abe Mitsuki');
INSERT INTO `abc` (`id`, `name`) VALUES (9, 'Du Shihan');
INSERT INTO `abc` (`id`, `name`) VALUES (10, 'Chiang Chi Yuen');
COMMIT;
BEGIN;
INSERT INTO `table_0` (`id`, `name`) VALUES (1, 'Luo Rui');
INSERT INTO `table_0` (`id`, `name`) VALUES (2, 'Yung Wing Kuen');
INSERT INTO `table_0` (`id`, `name`) VALUES (3, 'Chiang Chun Yu');
INSERT INTO `table_0` (`id`, `name`) VALUES (4, 'Tang Ming');
INSERT INTO `table_0` (`id`, `name`) VALUES (5, 'Man Wai Lam');
INSERT INTO `table_0` (`id`, `name`) VALUES (6, 'Tin Tsz Ching');
INSERT INTO `table_0` (`id`, `name`) VALUES (7, 'Doris Moore');
INSERT INTO `table_0` (`id`, `name`) VALUES (8, 'Abe Mitsuki');
INSERT INTO `table_0` (`id`, `name`) VALUES (9, 'Du Shihan');
INSERT INTO `table_0` (`id`, `name`) VALUES (10, 'Chiang Chi Yuen');
COMMIT;
INSERT INTO `table_1` (`id`, `name`) VALUES (100, 'tom');

配置容器路由转发

我们在代码中开发过程中可能会用到容器的ip地址,例如上面的172.20.80.0/24这个网段,但是你会发现你是ping不通的,这里设计到了一些docker网络的一些知识,可以在网上看一下资料,这里只给出解决方法
安装路由转发镜像

# 现在连接器
brew install wenjunxiao/brew/docker-connector
# 加入路由
docker network ls --filter driver=bridge --format "{{.ID}}" | xargs docker network inspect --format "route {{range .IPAM.Config}}{{.Subnet}}{{end}}" >> /opt/homebrew/etc/docker-connector.conf
# 启动路由器
sudo /opt/homebrew/opt/docker-connector/bin/docker-connector -config /opt/homebrew/etc/docker-connector.conf
# 启动镜像
docker run -it -d --restart always --net host --cap-add NET_ADMIN --name connector wenjunxiao/mac-docker-connector

如果还是ping不通就重启一下上面的转发容器
这一步很重要,想要通过访问容器的ip就要完成这一步

依赖包准备

下载flinkcdc 的依赖包放到flink目录下并解压
flinkcdc 依赖 : flink-cdc-3.0.0-bin.tar.gz
下载连接器 的依赖包放到flinkcdc的lib目录下
connector 依赖 :

  • MySQL pipeline connector 3.0.0
  • Apache Doris pipeline connector 3.0.0

配置FLINK_HOME环境变量

pwd
/Users/antg/software/flink-1.18.0/
export FLINK_HOME=/Users/antg/software/flink-1.18.0/

数据同步

整库同步

编写yaml文件 mysql-to-doris.yaml

################################################################################
# Description: Sync MySQL all tables to Doris
################################################################################
source:type: mysqlhostname: localhostport: 3306username: rootpassword: 12345678tables: doris_sync.\.*server-id: 5400-5404server-time-zone: Asia/Shanghaisink:type: dorisfenodes: 127.0.0.1:8031username: rootpassword: ""table.create.properties.light_schema_change: truetable.create.properties.replication_num: 1pipeline:name: Sync MySQL Database to Dorisparallelism: 2

启动任务

bash bin/flink-cdc.sh mysql-to-doris.yaml

查看页面效果image.png
这里可以看到同步的数据条数及大小

查看doris的数据及建表情况
image.png
可以看到表被自动创建并且数据也同步过来了

新增数据

INSERT INTO `a_0` (`id`, `name`) VALUES (3, 'jack');

Kapture 2024-01-21 at 15.04.36.gif

更新数据

update a_0 set name='tom' where id=3;

Kapture 2024-01-21 at 15.09.41.gif

删除数据

delete from a_0 where id=1;

没成功同步(已咨询社区是1.2.2的bug,在1.2.3修复了,正常来说会同步)

新增字段

alter table a_0 add column age int;

在这里插入图片描述

修改字段

# 修改名称
alter table a_0 change age age_range int;
# 修改字段类型
alter table a_0 modify column age_range varchar(100);
# 字段字段长度
alter table a_0 modify column age_range varchar(1200);

以上语句不会被同步

删除字段

alter table a_0 drop column age_range;

以上语句不会被同步

删除表

drop table a_0;

不会被同步

结论 :
1.新增数据,新增字段,修改数据会被实时同步到doris
2.delete数据不会被同步(已咨询社区是1.2.2的bug,在1.2.3修复了,正常来说会同步)
3.修改字段名称,类型,长度不会被同步(可能有参数可以开启)
4.删除字段不会被同步
5.删除表不会被同步

路由变更

这里将使用flinkcdc3.0 新增的路由功能来实现分表合一的效果,而且也可以做到同步到doris的库名和表名换成自己想要的名称
将之前的mysql端数据清理,表重新建立

需求 :
将mysql端doris_sync同步到doris的ods库中
a_0,a_1 合并到ods_a表
abc 同步到 ods_abc表
table_0,table_1同步到 ods_table表

任务配置 route.yaml

################################################################################
# Description: Sync MySQL all tables to Doris
################################################################################
source:type: mysqlhostname: localhostport: 3306username: rootpassword: 12345678tables: doris_sync.\.*server-id: 5400-5404server-time-zone: Asia/Shanghaisink:type: dorisfenodes: 127.0.0.1:8031username: rootpassword: ""table.create.properties.light_schema_change: truetable.create.properties.replication_num: 1route:- source-table: doris_sync.a_\.*sink-table: ods.ods_a- source-table: doris_sync.abcsink-table: ods.ods_abc- source-table: doris_sync.table_\.*sink-table: ods.ods_tablepipeline:name: Sync MySQL Database to Dorisparallelism: 2

创建doris端ods库(不会自动创建库,必须手动创建)

create database ods;

将之前的任务停掉,启动这个任务
image.png
在这里插入图片描述

可以看到
1.多个分表在doris只创建了一个目标表
2.多个分表的数据都同步到了一个表中
非常棒的功能 👍👍👍

测试一下新增一个分表是否会自动同步到目标表

CREATE TABLE `a_2` (`id` int NOT NULL AUTO_INCREMENT,`name` varchar(255) DEFAULT NULL,PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;INSERT INTO `a_2` (`id`, `name`) VALUES (1000, 'a');

image.png
新增分表后,分表不会被自动同步

重启任务
image.png
重启后数据可以被正常同步

从checkpoint恢复任务并新增分表

先修改一下flink-conf.yaml,否则任务cancel的时候ck不会被保留,还需要修改一下ck存储的路径

# 在flink目录下创建一个路径存储ck
mkdir ckdata

image.png
image.png

启动任务

bash bin/flink-cdc.sh route.yaml

image.png
看一下ck是否正常存储
image.png
新增表,cancel任务,然后从ck处重启

CREATE TABLE `a_4` (`id` int NOT NULL AUTO_INCREMENT,`name` varchar(255) DEFAULT NULL,PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;INSERT INTO `a_4` (`id`, `name`) VALUES (1000000, 'a');

################################################################################
# Description: Sync MySQL all tables to Doris
################################################################################
source:type: mysqlhostname: localhostport: 3306username: rootpassword: 12345678tables: doris_sync.\.*server-id: 5400-5404server-time-zone: Asia/Shanghaisink:type: dorisfenodes: 127.0.0.1:8031username: rootpassword: ""table.create.properties.light_schema_change: truetable.create.properties.replication_num: 1route:- source-table: doris_sync.a_\.*sink-table: ods.ods_a- source-table: doris_sync.abcsink-table: ods.ods_abc- source-table: doris_sync.table_\.*sink-table: ods.ods_tablepipeline:name: Sync MySQL Database to Dorisparallelism: 2

在flink-conf最后加上ck的重启路径

# 查看当前路径
pwd
/Users/antg/software/flink-1.18.0/flink-cdc-3.0.0# 找到最新的ck存储路径
ll -rth ../ckdata
drwxr-xr-x@ 5 antg  staff   160B Jan 21 16:27 436dfeb839b2c877d6e49023e3e099b5
drwxr-xr-x@ 5 antg  staff   160B Jan 21 17:12 d519a3f930d9f410e048f63a883e1dce
drwxr-xr-x@ 5 antg  staff   160B Jan 21 18:59 b0ed22a804ad34336ab3e9b328d13257
drwxr-xr-x@ 5 antg  staff   160B Jan 21 19:01 394d7a89885bbd319e8ab92043283de9
drwxr-xr-x@ 5 antg  staff   160B Jan 21 19:05 1547d3cf60ed278ccd3787025bb4b5f6
drwxr-xr-x@ 5 antg  staff   160B Jan 21 19:07 51ff313e98fb9882f20f57bc697a8ae6
drwxr-xr-x@ 5 antg  staff   160B Jan 21 19:08 f10623b642135002499775274c078b9e
drwxr-xr-x@ 5 antg  staff   160B Jan 21 19:09 73b47091ca00547a5d8121474b3dbd79ll ../ckdata/73b47091ca00547a5d8121474b3dbd79
drwxr-xr-x@ 3 antg  staff    96B Jan 21 19:09 chk-172
drwxr-xr-x@ 2 antg  staff    64B Jan 21 19:09 shared
drwxr-xr-x@ 2 antg  staff    64B Jan 21 19:09 taskowned# 将ck路径加到flink-conf的最后一行
vim ../conf/flink-conf.yaml
execution.savepoint.path: file:///Users/antg/software/flink-1.18.0/ckdata/73b47091ca00547a5d8121474b3dbd79/chk-172# 启动任务
bin/flink-cdc.sh route.yaml

image.png
可以看到任务从检查点重启了
image.png
数据也正常同步

这里从ck重启是修改了flink-conf,但是感觉这样很不方便,尝试过在yaml的pipeline下加上这个属性,但是不起作用,其他位置也没找到加ck路径的地方,如果各位大神有其他好的方法欢迎评论区留言,也欢迎加我的个人微信一起交流各种技术.

参考

[基于 Flink CDC 3.0 构建 MySQL 到 Doris 的 Streaming ELT] : https://ververica.github.io/flink-cdc-connectors/release-3.0/content/%E5%BF%AB%E9%80%9F%E4%B8%8A%E6%89%8B/mysql-doris-pipeline-tutorial-zh.html
[vm.max_map_count参数详解] : https://blog.csdn.net/a772304419/article/details/132585239

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

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

相关文章

YOLOv8改进 | 主干篇 | 低照度图像增强网络SCINet改进黑暗目标检测（全网独家首发）

YOLOv8改进 | 主干篇 | 低照度图像增强网络SCINet改进黑暗目标检测（全网独家首发）

一、本文介绍 本文给大家带来的改进机制是低照度图像增强网络SCINet，SCINet（自校正照明网络）是一种专为低光照图像增强设计的框架。它通过级联照明学习过程和权重共享机制来处理图像，优化了照明部分以提升图像质量。我将该网络集成在YOLOv8的主干上针对于图像的输入进行增…
阅读更多...
网上训练补题

网上训练补题

1.Codeforces Round 920 (Div. 3) A. 简单判断出边长即可 #include <stdio.h> #include <stdlib.h> #include <math.h> int main() {int n;scanf("%d", &n);while (n--){int x, y;scanf("%d%d", &x, &y);int e, f;scanf(&q…
阅读更多...
SQL 注入总结（详细）

SQL 注入总结（详细）

一、前言 这篇文章是最近学习 SQL 注入后的笔记&#xff0c;里面整理了 SQL 常见的注入方式&#xff0c;供大家学习了解 SQL 注入的原理及方法&#xff0c;也方便后续自己回顾&#xff0c;如有什么错误的地方欢迎指出&#xff01; 二、判断注入类型 按照注入点类型分类 数字型…
阅读更多...
【Coding】寒假每日一题Day.5. 互质数的个数

【Coding】寒假每日一题Day.5. 互质数的个数

题目来源 题目来自于AcWing平台&#xff1a;https://www.acwing.com/problem/content/4971/ 以blog的形式记录程序设计算法学习的过程&#xff0c;仅做学习记录之用。 题目描述 输入输出格式与数据范围 样例 第一组&#xff1a; 2 516第二组&#xff1a; 12 711943936思路…
阅读更多...
SpringMVC(八）处理AJAX请求

SpringMVC(八）处理AJAX请求

一、处理AJAX之准备工作： 首先我们创建一个新的工程： 我们将pom.xml复制过来: <?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-in…
阅读更多...
Excel乱码？教你3个简单解决方法！

Excel乱码？教你3个简单解决方法！

“我在编辑一个文件时&#xff0c;Excel突然就乱码了&#xff0c;怎么会这样呢&#xff1f;这个文件对我来说是比较重要的&#xff01;有什么方法可以快速解决吗&#xff1f;” 在处理Excel文件时&#xff0c;我们有时会遇到乱码问题。乱码不仅影响数据的可读性&#xff0c;还可…
阅读更多...
华清远见作业第三十一天——网络编程（第六天）

华清远见作业第三十一天——网络编程（第六天）

思维导图&#xff1a; 代码&#xff1a; 服务器&#xff1a; #include<stdio.h> #include<string.h> #include<stdlib.h> #include <sys/types.h> #include <sys/stat.h> #include <a.h> #define SER_PORT 10000 #define login 1 //登录…
阅读更多...
《WebKit 技术内幕》之七（4）： 渲染基础

《WebKit 技术内幕》之七（4）： 渲染基础

4　WebKit软件渲染技术 4.1　软件渲染过程 在很多情况下&#xff0c;也就是没有那些需要硬件加速内容的时候&#xff08;包括但不限于CSS3 3D变形、CSS3 03D变换、WebGL和视频&#xff09;&#xff0c;WebKit可以使用软件渲染技术来完成页面的绘制工作&#xff08;除非读者强行…
阅读更多...
【Redis】Redis如何实现key的过期删除

【Redis】Redis如何实现key的过期删除

​ &#x1f34e;个人博客&#xff1a;个人主页 &#x1f3c6;个人专栏&#xff1a;Redis ⛳️ 功不唐捐&#xff0c;玉汝于成 ​ 目录 前言 正文 结语 我的其他博客 前言 在当今信息时代&#xff0c;数据的快速存储和高效检索成为了软件系统设计中的核心需求。Redis作为…
阅读更多...
为什么PostgreSQL如此受追捧？

为什么PostgreSQL如此受追捧？

今天在群里看到一张图&#xff0c; 题目叫为什么PostgreSQL如此受追捧&#xff1f;列举了PG的7大特性&#xff0c;感觉很有意思&#xff0c;实际上在中国PG并没有Mysql流行&#xff0c;但是mysql被oracle收购之后&#xff0c;大家对mysql的发展前景不免有些担心&#xff0c;但是…
阅读更多...
golang学习笔记——http.Handle和http.HandleFunc的区别与type func巧妙运用

golang学习笔记——http.Handle和http.HandleFunc的区别与type func巧妙运用

文章目录 http.Handle和http.HandleFunc的区别http.Handle分析type func巧妙运用 http.HandleFunc分析总结参考资料 http.Handle和http.HandleFunc的区别 http.Handle和http.HandleFunc的区别体现了Go语言接口的巧妙运用 下面代码启动了一个 http 服务器&#xff0c;监听 808…
阅读更多...
系统引导器GRUB

系统引导器GRUB

全称为GNU GRUB&#xff0c;来自GNU计划的多操作系统引导器。 作用&#xff1a; 查看引导分区内容 ls /boot ls /boot/grub2/ 查看GRUB2的配置文件 cat /boot/grub2/grub.cfg 可以看到/boot中保存的文件主要是linux内核、内存映像文件等。 注意到是子用户&#xff0c;但在…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023