一键实现 Oracle 数据整库同步至 Apache Doris

在实时数据仓库建设或迁移的过程中,用户必须考虑如何高效便捷将关系数据库数据同步到实时数仓中来,Apache Doris 用户也面临这样的挑战。而对于从 Oracle 到 Doris 的数据同步,通常会用到以下两种常见的同步方式:

OGG/XStream/LogMiner 工具: 通过该方式先将数据同步到 Kafka 中,然后通过 Routine Load 消费 Kafka 中的数据进行实时同步。这种方式的同步链路相对较长,特别是在上游数据表较多的情况下,需要手动创建大量的 Routine Load 作业,同步流程不仅繁琐,也给用户增加了较大的使用及维护压力。

FlinkCDC: 该方式虽然可以直接将上游数据同步到 Doris 中,并在一定程度上缩短了同步链路,实际在使用过程中还会遇到以下问题:

  • 数据同步时,需要在 Flink 中对每张表手动配置参数及字段映射,尤其是在多表或整库同步场景中,不仅带来大量配置工作量,还增加了 FlinkSQL 脚本的维护成本。
  • 数据同步时,需要事先在 Doris 中手动逐个创建表,而面对数量庞大的上游表时,手动创建表不仅耗费时间,而且工作效率很低,间接影响数据同步的效率。
  • 由于每张 Source 表都会使用同一个链接,因此在整库同步时会给源端造成很大的链接压力。

为了解决上述问题,在新版本的 Doris-Flink-Connector  中,我们实现了 FlinkCDC 的 Datastream API 集成,无需提前在 Doris 中创建表以及映射关系,仅仅通过简单的参数配置就能一键完成从 Oracle 等关系型数据库到 Apache Doris 的整库数据同步。

此外,Doris-Flink-Connector 也可以一键实现万表 MySQL 整库同步至 Apache Doris 中来,具体使用可参考:一键实现万表 MySQL 整库同步至 Apache Doris

同步流程 & 实战演示


在进行整库同步前,我们先了解一下具体同步流程:

oracle.png

  • 在启动 Flink 任务之前,Doris-Flink-Connector  会自动读取需要同步的 Oracle 表的元数据信息,并自动在 Doris 中创建相应的表。
  • 通过 FlinkCDC 提供的 OracleSource 功能,能够从 Oracle 数据库中读取数据,并将其传递到下游进行处理。
  • 通过 Flink 的侧输出流功能,根据自定义规则将数据分流到不同的 Doris Sink 中,并同步到 Doris 中来。

通过以上简单操作,即可实现上游 Oracle 数据库的整库数据实时数据接入到 Apache Doris 中。接下来我们通过一个实际案例来详细说明具体的操作步骤:

01  Oracle 环境准备

# 拉取镜像
docker pull registry.cn-hangzhou.aliyuncs.com/helowin/oracle_11g# 启动镜像
docker run -it -d \
--privileged \
-p 1521:1521 \
--name oracle11g \
-e ORACLE_ALLOW_REMOTE=true \
-v /mnt/disk1/oracle:/data/oracle \
registry.cn-hangzhou.aliyuncs.com/helowin/oracle_11g# 进入容器
docker exec -it oracle11g bash

Oracle 归档日志(Binlog)配置:启动归档日志时,需对日志大小和存放地址进行设置,设置完成需进行重启。该步骤完成后才可进行后续增量数据的同步。

# 进入SQL命令行
[oracle@ef6d9de18e59 ~]$ sqlplus /nolog
SQL> conn /as sysdba
Connected.SQL> alter system set db_recovery_file_dest_size = 10G;
System altered.SQL> alter system set db_recovery_file_dest = '/home/oracle/oracle-data' scope=spfile;
System altered.SQL> shutdown immediate;
Database closed.
Database dismounted.
ORACLE instance shut down.SQL> startup mount;
ORACLE instance started.
Total System Global Area 1603411968 bytes
Fixed Size                  2213776 bytes
Variable Size             402655344 bytes
Database Buffers         1174405120 bytes
Redo Buffers               24137728 bytes
Database mounted.SQL> alter database archivelog;
Database altered.SQL> alter database open;
Database altered.
# 检查日志归档是否开启
SQL> archive log list;
Database log mode              Archive Mode
Automatic archival             Enabled
Archive destination            USE_DB_RECOVERY_FILE_DEST
Oldest online log sequence     1
Next log sequence to archive   1
Current log sequence           1# 启用补充日志记录
SQL> ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;
Database altered.#创建用户
CREATE USER admin IDENTIFIED BY admin123;
GRANT dba TO admin;

数据准备

[oracle@ef6d9de18e59 ~]$ sqlplus admin/admin123 
SQL> CREATE TABLE PERSONS(ID NUMBER(10),NAME VARCHAR2(128) NOT NULL,PRIMARY KEY(ID));
Table created.SQL> INSERT INTO "PERSONS" VALUES (1, 'zhangsan');
SQL> INSERT INTO "PERSONS" VALUES (2, 'lisi');
SQL> INSERT INTO "PERSONS" VALUES (3, 'wangwu');SQL> CREATE TABLE PERSONS_1(ID NUMBER(10),NAME VARCHAR2(128) NOT NULL,PRIMARY KEY(ID));
Table created.SQL> INSERT INTO "PERSONS_1" VALUES (1, 'zhangsan');
SQL> INSERT INTO "PERSONS_1" VALUES (2, 'lisi');
SQL> INSERT INTO "PERSONS_1" VALUES (3, 'wangwu');

02  Flink 环境配置

将 FlinkCDC-Oracle 的依赖和 Doris-Flink-Connector 包放到 Flink 的 lib 目录下,同时启动 Flink 集群。

# 下载相关依赖
wget https://repo.maven.apache.org/maven2/com/ververica/flink-sql-connector-oracle-cdc/2.3.0/flink-sql-connector-oracle-cdc-2.3.0.jar
wget https://repository.apache.org/content/repositories/snapshots/org/apache/doris/flink-doris-connector-1.16/1.5.0-SNAPSHOT/flink-doris-connector-1.16-1.5.0-20230811.065053-1.jar -O flink-doris-connector-1.16-1.5.0-SNAPSHOT.jar# 启动Flink集群
bin/start-cluster.sh

03  一键提交整库同步作业

本次同步以 PERSON 开头的所有的表。

<FLINK_HOME>/bin/flink run \-Dexecution.checkpointing.interval=10s \-Dparallelism.default=1 \-c org.apache.doris.flink.tools.cdc.CdcTools \./lib/flink-doris-connector-1.16-1.5.0-SNAPSHOT.jar \oracle-sync-database \--database test_db \--oracle-conf hostname=127.0.0.1 \--oracle-conf port=1521 \--oracle-conf username=admin \--oracle-conf password=admin123 \--oracle-conf database-name=HELOWIN \--oracle-conf schema-name=ADMIN \--including-tables "PERSONS.*" \--sink-conf fenodes=127.0.0.1:8030 \--sink-conf username=root \--sink-conf password=\--sink-conf jdbc-url=jdbc:mysql://127.0.0.1:9030 \--sink-conf sink.label-prefix=label \--table-conf replication_num=1

详细参数可参考:https://doris.apache.org/zh-CN/docs/dev/ecosystem/flink-doris-connector

提交成功后,可以在 FlinkWeb 上看到该同步任务的状态。

oracle2.png

进入 Doris 可以查看自动创建的表以及同步成功的全量数据。

mysql> use test_db;                                                                                                                                        
Reading table information for completion of table and column names                                                                                         
You can turn off this feature to get a quicker startup with -A                                                                                             Database changed                                                                                                                                           
mysql> show tables;                                                                                                                                        
+-------------------+                                                                                                                                      
| Tables_in_test_db |                                                                                                                                      
+-------------------+                                                                                                                                      
| PERSONS           |                                                                                                                                      
| PERSONS_1         |                                                                                                                                      
+-------------------+                                                                                                                                      
2 rows in set (0.00 sec)                                                                                                                                   mysql> select * from PERSONS;                                                                                                                              
+------+----------+                                                                                                                                        
| ID   | NAME     |                                                                                                                                        
+------+----------+                                                                                                                                        
|    2 | lisi     |                                                                                                                                        
|    3 | wangwu   |                                                                                                                                        
|    1 | zhangsan |                                                                                                                                        
+------+----------+                                                                                                                                        
3 rows in set (0.01 sec)                                                                                                                                   mysql> select * from PERSONS_1;                                                                                                                            
+------+----------+                                                                                                                                        
| ID   | NAME     |                                                                                                                                        
+------+----------+                                                                                                                                        
|    2 | lisi     |                                                                                                                                        
|    3 | wangwu   |                                                                                                                                        
|    1 | zhangsan |                                                                                                                                        
+------+----------+                                                                                                                                        
3 rows in set (0.01 sec)

在 Oracle 中模拟实时增删改数据

INSERT INTO PERSONS VALUES(4,'doris');
UPDATE PERSONS SET name = 'zhangsan-update' WHERE ID =1;
DELETE PERSONS WHERE ID =2; 

在 Doris 中进行验证,可以确认增量数据已经成功同步。

mysql> select * from PERSONS;                                                                                                                              
+------+-----------------+                                                                                                                                 
| ID   | NAME            |                                                                                                                                 
+------+-----------------+                                                                                                                                 
|    1 | zhangsan-update |                                                                                                                                 
|    4 | doris           |                                                                                                                                 
|    3 | wangwu          |                                                                                                                                 
+------+-----------------+                                                                                                                                 
3 rows in set (0.01 sec)  

通过以上操作,成功实现将 Oracle 中数据整库同步到 Doris 中,同时也实现了上游全量与增量数据的自动接入。

实际使用反馈


原先将 Oracle 数据同步到 Doris 中时,需要手动创建 Source 和 Sink 表,而使用 Doris-Flink-Connector 后可以实现多表、整库数据一键同步,极大简化了开发流程,该工具还能实现字段类型自动转换,数据同步更加简单便捷。

—— 远景动力 资深大数据工程师 孙全隆

在使用 Doris-Flink-Connector 之前,我们一般是通过 DataX 定时从业务系统中抽取数据,当进行全量同步时,抽取数据会对业务系统造成一定的压力,且该方式只能做到小时级的同步。期间我们也尝试了 FlinkCDC,该方式虽然可以实现数据实时写入 Doris ,但每个表都需要手动创建新任务,配置工作量大且会浪费服务器资源。而 Doris-Flink-Connector 可以实现一键化脚本操作,为我们减少了繁杂的手工配置流程,高效稳定的实现了整库数据快速同步。

—— 郑煤机数耘科技 资深大数据工程师 杨开元

Doris-Flink-Connector  一键操作即可快速实现 Oracle 数据整库同步到 Doris,节省了手动配置以及编写复杂同步代码的步骤,避免了手动同步中可能出现数据不一致的问题。不仅能提高数据的准确性和可靠性,也极大提升了工作的效率。

—— 海程邦达 资深大数据工程师 王新

在实时数仓的建设过程中,对于 ODS 贴源数据层的同步需求,Doris-Flink-Connector  能够很好的解决全量数据、增量数据、增量表、表结构变更自动监听。同时它也对 Stream Load 逻辑进行了优化,可以避免频繁对空数据进行 Load,减轻了数据库压力。此外,Doris-Flink-Connector 能够帮助我们节省大量 Flink 集群资源,特别是业务变更频繁时期,能很好及时的同步上游状态,确保上下游数据的一致性。

——旺小宝 数据架构师 米华军

我们在 MySQL 和 Orcale 两个场景下均进行了全量 + 增量的尝试,Doris-Flink-Connector 是真正的拆箱即用,真正实现了一键式操作、无感知建表,这为开发人员节省了不少时间成本,同时在使用期间遇到问题,SelectDB 技术同学的响应速度非常给力,帮助我们快速推进数据同步工作。

—— 博思软件 资深大数据开发工程师 刘工

总结


Doris-Flink-Connector 通过集成 FlinkCDC,能够将上游 Oracle 数据库中的数据快速同步到 Doris 中。特别是在整库同步场景中,用户只需执行一键导入命令,即可快速将整个数据库的全量和增量数据导入到 Doris 中。这一功能的引入大大降低了数据同步的门槛,使数据同步变得更加简单高效。

最后,欢迎有需要的小伙伴使用该工具,感兴趣的伙伴可以在评论区留言或私信申请进入专项支持群,如果你在使用过程中遇到任何问题,均可向我们反馈~

# 作者介绍: 吴迪, SelectDB 生态研发工程师。

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

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

相关文章

设计模式-桥接模式

核心思想 适配器模式类似&#xff0c;以后也会遇到意思接近一样的设计模式。在开发中一般多个模式混用&#xff0c;且根据不同的场景进行搭配&#xff0c;桥接模式也是结构型模式将抽象的部分和实现的部分分离&#xff0c;使它们都可以独立的变化。通俗来说&#xff0c;就是通…

Linux特殊指令

目录 1.dd命令 2.mkfs格式化 3.df命令 4.mount实现硬盘的挂载 5.unshare 1.dd命令 dd命令可以用来读取转换并输出数据。 示例一&#xff1a; if表示infile&#xff0c;of表示outfile。这里的/dev/zero是一个特殊文件&#xff0c;会不断产生空白数据。 bs表示复制一块的大…

vscode 对模型train、detect脚本进行Debug时配置参数

我们训练yolov5代码时&#xff0c;一般会配置一些参数&#xff0c;比如模型权重文件--weights, 模型的配置文件--cfg, 以及训练的数据--data, 对应的训练脚本为: 训练train python train.py -- weights ./yolov5s.pt --cfg models\yolov5s.yaml --data ./data/coco128.yaml…

9.oracle中sign函数

在Oracle/PLSQL中, sign 函数返回一个数字的正负标志. 语法如下&#xff1a;sign( number ) number 要测试标志的数字. If number < 0, then sign returns -1. If number 0, then sign returns 0. If number > 0, then sign returns 1. 应用于: Oracle 8i, Oracle …

Docker - Docker安装MySql并启动

因为项目需要连接数据库&#xff0c;但是远程服务器上的mysql我不知道账户和密码&#xff0c;这个时候便是docker发挥作用的关键时刻了&#xff01; 目录 docker安装安装gcc卸载老docker&#xff08;如有&#xff09;安装软件包设置镜像仓库更新yum软件包索引安装docker启动doc…

SSM框架的学习与应用(Spring + Spring MVC + MyBatis)-Java EE企业级应用开发学习记录(第五天)MyBatis的注解开发

SSM框架的学习与应用(Spring Spring MVC MyBatis)-Java EE企业级应用开发学习记录&#xff08;第五天&#xff09;MyBatis的注解开发 ​ 昨天我们深入学习了MyBatis多表之间的关联映射&#xff0c;了解掌握了一对一关联映射&#xff0c;一对多关联映射&#xff0c;嵌套查询方…

微信小程序如何实现页面传参和页面传递多个参数

前言 只要你的小程序超过一个页面那么可能会需要涉及到页面参数的传递&#xff0c;下面我总结了 4 种页面方法。 下面时多个参数页面传参的方式 let loveJSON.stringify(this.data.totle);let youJSON.stringify(this.data.totleId)let csdnJSON.stringify(this.data.totleP…

Linux系统下vim常用命令

一、基础命令&#xff1a; v:可视模式 i:插入模式 esc:命令模式下 :q &#xff1a;退出 :wq &#xff1a;保存并退出 ZZ&#xff1a;保存并退出 :q! &#xff1a;不保存并强制退出二、在Esc下&#xff1a; dd : 删除当前行 yy:复制当前行 p:复制已粘贴的文本 u:撤销上一步 U:…

ELK日志收集系统

目录 一、概述 二、组件 一、logstash 一、工作过程 二、INPUT 三、FILETER 四、OUTPUTS 二、elasticsearch 三、kibana 三、架构类型 一、ELK 二、ELKK 三、ELFK 四、ELFKK 五、EFK 四、配置ELK日志收集系统集群实验的步骤文档 五、配置ELK日志收集系统集群 …

图像二值化

目录 1.固定值二值化 2.自适应阈值二值化 3.Android JNI完整代码 1.固定值二值化 固定阈值二值化是OpenCV中一种简单而常用的图像处理技术&#xff0c;用于将图像转换为二值图像。在固定阈值二值化中&#xff0c;像素值根据一个预定义的阈值进行分类&#xff0c;大于阈值的…

【JavaEE】Spring事务-@Transactional参数介绍-事务的隔离级别以及传播机制

【JavaEE】Spring 事务&#xff08;2&#xff09; 文章目录 【JavaEE】Spring 事务&#xff08;2&#xff09;1. Transactional 参数介绍1.1 value 和 transactionManager1.2 timeout1.3 readOnly1.4 后面四个1.5 isolation 与 propagation 2. Spring 事务隔离级别 - isolation…

利用torchvision库实现目标检测与语义分割

一、介绍 利用torchvision库实现目标检测与语义分割。 二、代码 1、目标检测 from PIL import Image import matplotlib.pyplot as plt import torchvision.transforms as T import torchvision import numpy as np import cv2 import randomCOCO_INSTANCE_CATEGORY_NAMES …

SQL server数据库-定制查询-指定查询列/行、结果排序和Like模糊查询

本篇讲述进阶查询方法&#xff0c;如有语句不明确&#xff0c;可跳转本文专栏学习基础语法 1、指定列查询 特点 只会显示你输入的列的数据&#xff0c;会根据你输入的顺序进行显示&#xff0c;可以自定义查询显示时的列名 &#xff08;1&#xff09;只会显示你输入的列的数…

C++ 面试题(一)--C++基础,面向对象,内存管理

目录 1.part1 C基础 1 C特点 2 说说C语言和C的区别 3 说说 C中 struct 和 class 的区别 4 include头文件的顺序以及双引号""和尖括号<>的区别 5 说说C结构体和C结构体的区别 6 导入C函数的关键字是什么&#xff0c;C编译时和C有什么不同&#xff1f; 7…

系统架构设计高级技能 · 云原生架构设计理论与实践

系列文章目录 系统架构设计高级技能 软件架构概念、架构风格、ABSD、架构复用、DSSA&#xff08;一&#xff09;【系统架构设计师】 系统架构设计高级技能 系统质量属性与架构评估&#xff08;二&#xff09;【系统架构设计师】 系统架构设计高级技能 软件可靠性分析与设计…

新能源汽车动力总成系统及技术

需要动力系统总成的请联&#xff1a;shbinzer 拆车邦 需要动力系统总成的请联&#xff1a;shbinzer 拆车邦 需要动力系统总成的请联&#xff1a;shbinzer 拆车邦 需要动力系统总成的请联&#xff1a;shbinzer 拆车邦 需要动力系统总成的请联&#xff1a;shbinzer …

使用Python构建网络爬虫:提取网页内容和图片资源

网络爬虫是一种自动获取网页内容的程序&#xff0c;它可以帮助我们高效地收集网络上的有价值信息。本文将介绍如何使用Python构建网络爬虫&#xff0c;提取网页内容和图片资源。   一、环境准备   1.安装Python环境   首先&#xff0c;确保您已经安装了Python环境。访问P…

kettle实现爬虫

步骤概览 获取请求 请求地址 东方财富网股票请求 自定义常量数据 获取HTTP请求之前&#xff0c;必须先定义一个URL常量作为HTTP client的输入 HTTP client 注&#xff1a;此处得到的数据并不是原生的json字符串&#xff0c;自己可以用文本文件输出测试以下。如下图 JavaScri…

QT登陆注册界面练习

一、界面展示 二、主要功能界面代码 #include "widget.h" #include "ui_widget.h"Widget::Widget(QWidget *parent): QMainWindow(parent), ui(new Ui::Widget) {ui->setupUi(this);this->setFixedSize(540,410); //设置固定尺寸th…

android Junit4编写自测用例

10多年的android开发经验&#xff0c;一直以来呢&#xff0c;也没有使用过android自带的测试代码编写。说来也惭愧。今天也花了点时间稍微研究了下。还挺简单。接下来就简单的说一下。 新建工程 直接默认新建一个工程&#xff0c;就会有两个目录androidTest和test(unitTest)两…