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 下会包含 binliblogconf 四个目录。

  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

同样的,去修改 shipmentsproducts 表,也能在 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.order01app_db.order02app_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 多边形区域绘制组件封装

效果预览&#xff1a; CanvasBox组件 <!-- 区域设置canvas --> <template><div class"all" ref"divideBox"><!-- <div><button click"test">清空</button></div> --><img id"img"…

Ubuntu中MySQL远程登录设置

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

使用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;膨胀可…

使用 Logistic Regression 识别手写数字-PyTorch

逻辑回归是一种非常常用的统计方法,它允许我们从一组自变量预测二进制输出。本文之前已经介绍了 Logistic 回归的各种属性及其 Python 实现 。现在,我们将了解如何在 PyTorch 中实现这一点,PyTorch 是由 Facebook 开发的一个非常流行的深度学习库。 现在,我们将了解如何使用…

上门按摩系统开发方案源码搭建

上门按摩系统开发方案 一、项目概述 上门按摩系统是一个连接按摩技师和客户的平台&#xff0c;旨在提供便捷、高效的上门按摩服务。通过该系统&#xff0c;客户可以轻松预约合适的按摩技师&#xff0c;并享受个性化的按摩服务。 二、系统功能模块 用户管理模块&#xff1a;…

TCP——Socket

应用进程只借助Socket API发和收但是不关心他是怎么进行传和收的 数据结构 图示Socket连接 捆绑属于隐式捆绑

Redis非关系型数据库操作命令大全

以下是 Redis 的常用操作命令大全&#xff0c;涵盖了键值操作、字符串、哈希、列表、集合、有序集合、发布/订阅、事务等多个方面的操作。 1. 通用键命令 命令说明SET key value设置指定 key 的值GET key获取指定 key 的值DEL key删除指定的 keyEXISTS key检查 key 是否存在E…

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

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

Autosar软件组件概述

AUTOSAR中的应用软件设计为独立的软件单元&#xff0c;即软件组件类型&#xff08;SwComponentTypes&#xff09;&#xff0c;封装了相关功能和行为的实现&#xff0c;并通过端口原型&#xff08;PortPrototypes&#xff09;&#xff0c;向外部暴露自己。 AUTOSAR 标准应用了设…

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 …

【VUE】封装用户树形选择和部门树形选择控件

用vue实现封装用户树形选择和部门树形选择控件&#xff0c;采用el-tree。方便各个功能模块的使用和以后的开发。 一、封装用户树形选择控件&#xff08;userTree.vue&#xff09; <template><div style"padding: 10px;"><!-- 选择人员对话框 -->…

Java 类,变量,对象

类的声明 1、类的权限访问修饰符&#xff0c;public 缺省 public修涉的&#xff0c;在同包中可以进行访问&#xff0c;在不同包中也可以进行访问&#xff0c;缺省的修饰符只能在同包中进行访问 2、类的成员&#xff1a;属性&#xff0c;属性如何调用&#xff1f;对象.属性 1、权…

日常随笔1--MySQL中添加Redo日志文件的步骤详解

Redo日志是MySQL中重要的事务日志&#xff0c;用于记录事务对数据库进行的修改操作&#xff0c;确保数据库的持久性和恢复能力。本文将详细介绍在MySQL中添加Redo日志文件的步骤&#xff0c;包括设置Redo日志文件大小、位置和数量等。 1. 确认当前Redo日志文件状态 在添加新的…

xtu oj 原根

文章目录 回顾杂思路c 语言代码 回顾 AB III问题 H: 三角数问题 G: 3个数等式 数组下标查询&#xff0c;降低时间复杂度1405 问题 E: 世界杯xtu 数码串xtu oj 神经网络xtu oj 1167 逆序数&#xff08;大数据&#xff09; 杂 有一些题可能是往年的程设的题&#xff0c;现在搬到…

20255 - 中医方剂学 - 考研 - 执业

第1章 总论 1.我国现存最早的记载方剂的医书是&#xff08;&#xff09;( ) [单选] A.《太平圣惠方》 B.《黄帝内经》 C.《五十二病方》 D.《千金要方》 E.《外台秘要》 正确答案: C 2.我国最早的中医经典理论著作是&#xff08;&#xff09;( ) [单选] A.《伤寒杂病论…

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

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

VS code部署Vue项目Demo

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

9.指针与链表

一、指针 1.基本介绍 在程序中&#xff0c;我们的数据都有其存储的地址。在程序每次的实际运行过程中&#xff0c;变量在物理内存中的存储位置不尽相同。不过&#xff0c;我们仍能够在编程时&#xff0c;通过一定的语句&#xff0c;来取得数据在内存中的地址。地址也是数据。…