Flink CDC同步mysql数据到doris

前置参考

flink快速安装：Flink入门-CSDN博客

doris快速安装：Apache Doris快速安装-CSDN博客

Flink CDC 是一个基于流的数据集成工具，旨在为用户提供一套功能更加全面的编程接口（API）。 该工具使得用户能够以 YAML 配置文件的形式，优雅地定义其 ETL（Extract, Transform, Load）流程，并协助用户自动化生成定制化的 Flink 算子并且提交 Flink 作业。 Flink CDC 在任务提交过程中进行了优化，并且增加了一些高级特性，如表结构变更自动同步（Schema Evolution）、数据转换（Data Transformation）、整库同步（Full Database Synchronization）以及 精确一次（Exactly-once）语义。

Flink CDC 深度集成并由 Apache Flink 驱动，提供以下核心功能：

  • ✅ 端到端的数据集成框架
  • ✅ 为数据集成的用户提供了易于构建作业的 API
  • ✅ 支持在 Source 和 Sink 中处理多个表
  • ✅ 整库同步
  • ✅具备表结构变更自动同步的能力（Schema Evolution），

如何使用 Flink CDC

Flink CDC 提供了基于 YAML 格式的用户 API，更适合于数据集成场景。以下是一个 YAML 文件的示例，它定义了一个数据管道(Pipeline)，该Pipeline从 MySQL 捕获实时变更，并将它们同步到 Apache Doris：

source:type: mysqlhostname: localhostport: 3306username: rootpassword: 123456tables: app_db.\.*server-id: 5400-5404server-time-zone: UTCsink:type: dorisfenodes: 127.0.0.1:8030username: rootpassword: ""table.create.properties.light_schema_change: truetable.create.properties.replication_num: 1pipeline:name: Sync MySQL Database to Dorisparallelism: 2

通过使用 flink-cdc.sh 提交 YAML 文件，一个 Flink 作业将会被编译并部署到指定的 Flink 集群。 请参考 核心概念 以获取 Pipeline 支持的所有功能的完整文档说明。

Streaming ELT 同步 MySQL 到 Doris

这篇教程将展示如何基于 Flink CDC 快速构建 MySQL 到 Doris 的 Streaming ELT 作业，包含整库同步、表结构变更同步和分库分表同步的功能。 本教程的演示都将在 Flink CDC CLI 中进行，无需一行 Java/Scala 代码，也无需安装 IDE。

准备阶段

准备一台已经安装了 Docker 的 Linux 或者 MacOS 电脑。

  1. 下载 Flink 1.18.0，解压后得到 flink-1.18.0 目录。 使用下面的命令跳转至 Flink 目录下，并且设置 FLINK_HOME 为 flink-1.18.0 所在目录。

    cd flink-1.18.0

  2. 通过在 conf/flink-conf.yaml 配置文件追加下列参数开启 checkpoint，每隔 3 秒做一次 checkpoint。

    execution.checkpointing.interval: 3000

  3. 使用下面的命令启动 Flink 集群。

    ./bin/start-cluster.sh

启动成功的话，可以在 http://localhost:8081/访问到 Flink Web UI，如下所示：

Flink UI

多次执行 start-cluster.sh 可以拉起多个 TaskManager。

准备 Docker 环境

接下来的教程将以 docker-compose 的方式准备所需要的组件。

  1. 宿主机配置 由于 Doris 的运行需要内存映射支持，需在宿主机执行如下命令:

    sysctl -w vm.max_map_count=2000000

MacOS 由于内部实现容器的方式不同，在部署时宿主机直接修改max_map_count值可能无法成功，需要先创建以下容器：

docker run -it --privileged --pid=host --name=change_count debian nsenter -t 1 -m -u -n -i sh

容器创建成功执行以下命令：

sysctl -w vm.max_map_count=2000000

然后 exit 退出，创建 Doris Docker 集群。

  1. docker 镜像启动 使用下面的内容创建一个 docker-compose.yml 文件：

    version: '2.1'
services:doris:image: yagagagaga/doris-standaloneports:- "8030:8030"- "8040:8040"- "9030:9030"mysql:image: debezium/example-mysql:1.1ports:- "3306:3306"environment:- MYSQL_ROOT_PASSWORD=123456- MYSQL_USER=mysqluser- MYSQL_PASSWORD=mysqlpw

该 Docker Compose 中包含的容器有：

  • MySQL: 包含商品信息的数据库 app_db
  • Doris: 存储从 MySQL 中根据规则映射过来的结果表

在 docker-compose.yml 所在目录下执行下面的命令来启动本教程需要的组件：

docker-compose up -d

该命令将以 detached 模式自动启动 Docker Compose 配置中定义的所有容器。你可以通过 docker ps 来观察上述的容器是否正常启动了，也可以通过访问http://localhost:8030/ 来查看 Doris 是否运行正常。

在 MySQL 数据库中准备数据

  1. 进入 MySQL 容器

    docker-compose exec mysql mysql -uroot -p123456

  2. 创建数据库 app_db 和表 orders,products,shipments，并插入数据

    -- 创建数据库
CREATE DATABASE app_db;USE app_db;-- 创建 orders 表
CREATE TABLE `orders` (
`id` INT NOT NULL,
`price` DECIMAL(10,2) NOT NULL,
PRIMARY KEY (`id`)
);-- 插入数据
INSERT INTO `orders` (`id`, `price`) VALUES (1, 4.00);
INSERT INTO `orders` (`id`, `price`) VALUES (2, 100.00);-- 创建 shipments 表
CREATE TABLE `shipments` (
`id` INT NOT NULL,
`city` VARCHAR(255) NOT NULL,
PRIMARY KEY (`id`)
);-- 插入数据
INSERT INTO `shipments` (`id`, `city`) VALUES (1, 'beijing');
INSERT INTO `shipments` (`id`, `city`) VALUES (2, 'xian');-- 创建 products 表
CREATE TABLE `products` (
`id` INT NOT NULL,
`product` VARCHAR(255) NOT NULL,
PRIMARY KEY (`id`)
);-- 插入数据
INSERT INTO `products` (`id`, `product`) VALUES (1, 'Beer');
INSERT INTO `products` (`id`, `product`) VALUES (2, 'Cap');
INSERT INTO `products` (`id`, `product`) VALUES (3, 'Peanut');
Create database in Doris

Doris 暂时不支持自动创建数据库，需要先创建写入表对应的数据库。

  1. 进入 Doris Web UI。
    http://localhost:8030/
    默认的用户名为 root，默认密码为空。

    Doris UI

  2. 通过 Web UI 创建 app_db 数据库

    create database app_db;

    Doris create table

通过 FlinkCDC cli 提交任务

  1. 下载下面列出的二进制压缩包，并解压得到目录 flink cdc-3.0.0 '：
    flink-cdc-3.0.0-bin.tar.gz. flink-cdc-3.0.0 下会包含 bin、lib、log、conf 四个目录。

  2. 下载下面列出的 connector 包，并且移动到 lib 目录下 下载链接只对已发布的版本有效, SNAPSHOT 版本需要本地基于 master 或 release- 分支编译.

    • MySQL pipeline connector 3.0.0
    • Apache Doris pipeline connector 3.0.0

3.编写任务配置 yaml 文件 下面给出了一个整库同步的示例文件 mysql-to-doris.yaml：

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

其中： source 中的 tables: app_db.\.* 通过正则匹配同步 app_db 下的所有表。 sink 添加 table.create.properties.replication_num 参数是由于 Docker 镜像中只有一个 Doris BE 节点。

  1. 最后，通过命令行提交任务到 Flink Standalone cluster 
    bash bin/flink-cdc.sh mysql-to-doris.yaml

提交成功后，返回信息如：

Pipeline has been submitted to cluster.
Job ID: ae30f4580f1918bebf16752d4963dc54
Job Description: Sync MySQL Database to Doris

在 Flink Web UI，可以看到一个名为 Sync MySQL Database to Doris 的任务正在运行。

MySQL-to-Doris

打开 Doris 的 Web UI，可以看到数据表已经被创建出来，数据能成功写入。

Doris display data

同步变更

进入 MySQL 容器

 docker-compose exec mysql mysql -uroot -p123456

接下来，修改 MySQL 数据库中表的数据，Doris 中显示的订单数据也将实时更新：

  1. 在 MySQL 的 orders 表中插入一条数据

    INSERT INTO app_db.orders (id, price) VALUES (3, 100.00);

  2. 在 MySQL 的 orders 表中增加一个字段

    ALTER TABLE app_db.orders ADD amount varchar(100) NULL;

  3. 在 MySQL 的 orders 表中更新一条数据

    UPDATE app_db.orders SET price=100.00, amount=100.00 WHERE id=1;

  4. 在 MySQL 的 orders 表中删除一条数据

    DELETE FROM app_db.orders WHERE id=2;

每执行一步就刷新一次 Doris Web UI，可以看到 Doris 中显示的 orders 数据将实时更新，如下所示：

Doris display result

同样的，去修改 shipments, products 表，也能在 Doris 中实时看到同步变更的结果。

Route the changes

Flink CDC 提供了将源表的表结构/数据路由到其他表名的配置，借助这种能力，我们能够实现表名库名替换，整库同步等功能。 下面提供一个配置文件说明：

################################################################################
# Description: Sync MySQL all tables to Doris
################################################################################
source:type: mysqlhostname: localhostport: 3306username: rootpassword: 123456tables: app_db.\.*server-id: 5400-5404server-time-zone: UTCsink:type: dorisfenodes: 127.0.0.1:8030benodes: 127.0.0.1:8040username: rootpassword: ""table.create.properties.light_schema_change: truetable.create.properties.replication_num: 1route:- source-table: app_db.orderssink-table: ods_db.ods_orders- source-table: app_db.shipmentssink-table: ods_db.ods_shipments- source-table: app_db.productssink-table: ods_db.ods_productspipeline:name: Sync MySQL Database to Dorisparallelism: 2

通过上面的 route 配置，会将 app_db.orders 表的结构和数据同步到 ods_db.ods_orders 中。从而实现数据库迁移的功能。 特别地，source-table 支持正则表达式匹配多表，从而实现分库分表同步的功能，例如下面的配置：

route:- source-table: app_db.order\.*sink-table: ods_db.ods_orders

这样，就可以将诸如 app_db.order01、app_db.order02、app_db.order03 的表汇总到 ods_db.ods_orders 中。注意，目前还不支持多表中存在相同主键数据的场景，将在后续版本支持。

环境清理

本教程结束后，在 docker-compose.yml 文件所在的目录下执行如下命令停止所有容器：

docker-compose down

在 Flink 所在目录 flink-1.18.0 下执行如下命令停止 Flink 集群：

./bin/stop-cluster.sh

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

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

相关文章

洞察云上风险，主机安全尽在掌握

洞察云上风险，主机安全尽在掌握

在实战攻防演练中，主机一直是攻击方的最终目标。作为网络架构中的重要组成部分，主机包含了大量的敏感数据、关键服务和系统资源。同时主机拥有网络资源的访问权限，攻击者通过入侵主机获得权限，进而控制整个网络或系统。因此做好主…
阅读更多...
vue2 Canvas 多边形区域绘制组件封装

vue2 Canvas 多边形区域绘制组件封装

效果预览&#xff1a; CanvasBox组件 <!-- 区域设置canvas --> <template><div class"all" ref"divideBox"><!-- <div><button click"test">清空</button></div> --><img id"img"…
阅读更多...
Ubuntu中MySQL远程登录设置

Ubuntu中MySQL远程登录设置

mysql单独放在一台Ubuntu服务器上&#xff0c;我远程连接不上。可能是安装的时候忘记设置远程登录了。事后补救措施如下&#xff1a; MySQL 绑定地址配置问题 MySQL 可能只绑定了 localhost&#xff0c;无法接受来自外部主机的连接。你需要检查 MySQL 的配置文件 /etc/mysql/…
阅读更多...
使用Vscode配置ftp连接远程服务器（上传本地文件）

使用Vscode配置ftp连接远程服务器（上传本地文件）

1.安装插件 扩展商店搜sftp，点击进行安装。 2.配置json文件 crtl+shift+p 输入ftp配置命令 sftp:config {"name": "My Server", //设置名字"host": "localhost"</
阅读更多...
腐蚀膨胀预处理

腐蚀膨胀预处理

腐蚀&#xff1a;通过减少前景对象&#xff08;例如白色字符&#xff09;的边缘&#xff0c;腐蚀可以用来减小或消除细小的干扰线。如果干扰线较细&#xff0c;腐蚀可以有效地“消除”这些线条&#xff0c;同时保留较粗的字符。 膨胀&#xff1a;在腐蚀之后&#xff0c;膨胀可…
阅读更多...
TCP——Socket

TCP——Socket

应用进程只借助Socket API发和收但是不关心他是怎么进行传和收的 数据结构 图示Socket连接 捆绑属于隐式捆绑
阅读更多...
数据驱动时代：五款免费报表工具深度解析

数据驱动时代：五款免费报表工具深度解析

在当今数据驱动的时代&#xff0c;报表工具已经成为各类企业进行决策和管理的重要工具。无论是大中型企业还是小微企业&#xff0c;能够快速、高效地生成可视化报表&#xff0c;洞察业务运营情况&#xff0c;已经成为提升竞争力的关键。今天为大家挑选了5款非常优秀的报表软件&…
阅读更多...
Autosar软件组件概述

Autosar软件组件概述

AUTOSAR中的应用软件设计为独立的软件单元&#xff0c;即软件组件类型&#xff08;SwComponentTypes&#xff09;&#xff0c;封装了相关功能和行为的实现&#xff0c;并通过端口原型&#xff08;PortPrototypes&#xff09;&#xff0c;向外部暴露自己。 AUTOSAR 标准应用了设…
阅读更多...
std::setw中文不能对齐的问题

std::setw中文不能对齐的问题

从别的博主那 copy过来 记录下 #include <codecvt> #include <locale>class chs_codecvt : public std::codecvt_byname<wchar_t, char, std::mbstate_t> {public:chs_codecvt() : codecvt_byname("zh_CN.GBK") {} };int encoding_diff(string …
阅读更多...
xtu oj 原根

xtu oj 原根

文章目录 回顾杂思路c 语言代码 回顾 AB III问题 H: 三角数问题 G: 3个数等式 数组下标查询&#xff0c;降低时间复杂度1405 问题 E: 世界杯xtu 数码串xtu oj 神经网络xtu oj 1167 逆序数&#xff08;大数据&#xff09; 杂 有一些题可能是往年的程设的题&#xff0c;现在搬到…
阅读更多...
Higress 重磅更新：AI 能力全面开源，云原生能力再升级

Higress 重磅更新：AI 能力全面开源，云原生能力再升级

作者&#xff1a;澄潭、钰诚 新版本简介 Higress 最新的 1.4 版本基于为通义千问&#xff0c;以及多家云上 AGI 厂商客户提供 AI 网关的积累沉淀&#xff0c;开源了大量 AI 原生的网关能力。同时也在 Ingress、可观测、流控等云原生能力上做了全方位升级&#xff1a; AI 能力…
阅读更多...
VS code部署Vue项目Demo

VS code部署Vue项目Demo

在之前已经在IDEA中部署过vue项目demo。本次在上次基础上进行。 IDEA中Vue的安装和使用【window10】_idea安装vue-CSDN博客 步骤一、安装VSCode 双击安装即可 步骤二&#xff1a;检查npm是否安装 步骤三&#xff1a;检查vue是否安装 &#xff08;vue create 项目名 只要在v…
阅读更多...
Leetcode 单词规律

Leetcode 单词规律

即判断给定的模式字符串&#xff08;pattern&#xff09;和单词字符串&#xff08;s&#xff09;是否遵循相同的对应规则。具体来说&#xff0c;就是要判断 pattern 中的字符与 s 中的单词是否存在一一对应的关系&#xff0c;即双射&#xff08;bijection&#xff09;。 算法思…
阅读更多...
【解决办法】git clone报错unable to access ‘xxx‘: SSL certificate problem

【解决办法】git clone报错unable to access ‘xxx‘: SSL certificate problem

git clone 是 Git 版本控制系统中的一个基本命令&#xff0c;用于从远程仓库复制一个完整的版本库到本地。这个命令不仅复制远程仓库中的所有文件&#xff0c;还复制仓库的历史记录&#xff0c;使得你可以在本地进行版本控制操作&#xff0c;如提交&#xff08;commit&#xff…
阅读更多...
Electron+Vue实现两种方式的截屏功能

Electron+Vue实现两种方式的截屏功能

本次介绍的截屏功能一共有两种分别是在electron环境中与非electron环境中 非electron环境 这个环境下会有一些限制&#xff1a; 1.只能截浏览器中的画面 2.如果里面有iframe或者base64的图片会加载不出来&#xff08;这个会有解决办法&#xff09; yarn add -D js-web-scree…
阅读更多...
Java爬虫：获取商品评论数据的高效工具

Java爬虫：获取商品评论数据的高效工具

在电子商务的激烈竞争中&#xff0c;商品评论作为消费者购买决策的重要参考&#xff0c;对于商家来说具有极高的价值。它不仅能够帮助商家了解消费者的需求和反馈&#xff0c;还能作为改进产品和服务的依据。Java爬虫技术&#xff0c;以其稳健性和高效性&#xff0c;成为了获取…
阅读更多...
基于Spring Cloud的电商系统设计与实现——用户与商品模块的研究（上）

基于Spring Cloud的电商系统设计与实现——用户与商品模块的研究（上）

操作系统&#xff1a;Windows Java开发包&#xff1a;JDK1.8 项目管理工具&#xff1a;Maven3.6.0 项目开发工具&#xff1a;IntelliJIDEA 数据库&#xff1a;MySQL Spring Cloud版本&#xff1a;Finchley.SR2 Spring Boot版本&#xff1a;2.0.6.RELEASE 目录 用户模块—user-…
阅读更多...
YOLO系列入门：1、YOLO V11环境搭建

YOLO系列入门：1、YOLO V11环境搭建

YOLO了解 yolo检测原理 yolo是目标检测模型&#xff0c;目标检测包含物体分类、位置预测两个内容。目前yolo的开发公司官网为&#xff1a;https://docs.ultralytics.com/zh截止到目前2024年10月&#xff0c;最新的是yolo11。关于YOLO的介绍可以参考这篇文章&#xff1a;https…
阅读更多...
[Javase]封装、继承、多态与异常处理

[Javase]封装、继承、多态与异常处理

文章目录 一、前言二、封装1、封装的思想2、封装代码层面的体现 三、继承1、继承的概念和好处2、继承代码层面的体现 四、多态1、多态的概念2、多态的好处和三要素2、多态代码层面的体现 五、异常处理1、try-catch-finally结构详解2、throw\throws 一、前言 本文章适合有一定面…
阅读更多...
10.15.2024刷华为OD C题型（二）

10.15.2024刷华为OD C题型（二）

10.15.2024刷华为OD C题型&#xff08;二&#xff09; 密码输入检测智能成绩表 如果是目标院校150分能过&#xff0c;而且这道题是两百分的话我就阿弥陀佛了。 这类简单类型的字符串处理题目一看就有思路&#xff0c;起码能做&#xff0c;遇到那种稍微加点数学的&#xff0c;感…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023