一键实现 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 生态研发工程师。

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

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

【设计模式--原型模式(Prototype Pattern)

一、什么是原型模式 原型模式&#xff08;Prototype Pattern&#xff09;是一种创建型设计模式&#xff0c;它的主要目的是通过复制现有对象来创建新的对象&#xff0c;而无需显式地使用构造函数或工厂方法。这种模式允许我们创建一个可定制的原型对象&#xff0c;然后通过复制…

【Go 基础篇】Go 语言字符串函数详解:处理字符串进阶

大家好&#xff01;继续我们关于Go语言中字符串函数的探索。字符串是编程中常用的数据类型&#xff0c;而Go语言为我们提供了一系列实用的字符串函数&#xff0c;方便我们进行各种操作&#xff0c;如查找、截取、替换等。在上一篇博客的基础上&#xff0c;我们将继续介绍更多字…

React Hooks 全解:零基础入门

Hooks 的由来 你还在为该使用无状态组件&#xff08;Function&#xff09;还是有状态组件&#xff08;Class&#xff09;而烦恼吗&#xff1f; ——拥有了hooks&#xff0c;你再也不需要写Class了&#xff0c;你的所有组件都将是Function。 你还在为搞不清使用哪个生命周期钩…

相机成像之3A算法的综述

3A算法是摄像机成像控制技术中的三大自动控制算法。随着计算机视觉的迅速发展,该算法在摄像器材领域具有广泛的应用和前景。 那么3A控制算法又是指什么呢? (1)AE (Auto Exposure)自动曝光控制 (2)AF (Auto Focus)自动聚焦控制 (3)AWB (Auto White Balance)自动白平衡控…

使用Python统计小说语言描写的字数

说明&#xff1a;最早出现这个需求&#xff0c;来自博主阅读《罪与罚》&#xff0c;书中陀思妥耶夫斯基有太多的语言描述&#xff0c;以至于我想知道这本书中到底出现了多少对白。文本介绍如果使用python程序统计一本书中的对话&#xff0c;角色名称&#xff0c;标点符号。 找…

《Go 语言第一课》课程学习笔记(十一)

控制结构 if 的“快乐路径”原则 针对程序的分支结构&#xff0c;Go 提供了 if 和 switch-case 两种语句形式&#xff1b;而针对循环结构&#xff0c;Go 只保留了 for 这一种循环语句形式。 if 语句 if 语句是 Go 语言中提供的一种分支控制结构&#xff0c;它也是 Go 中最常…

华为手机实用功能介绍

一、内置app介绍 分四块介绍&#xff0c;包括出门款、规划款、工作款和生活款。 出门款&#xff1a;红色框框部分&#xff0c;照镜子化妆/看天气 规划款&#xff1a;黄色框框部分&#xff0c;日程表/计划表/番茄时间/计时 工作款&#xff1a;蓝色框框部分&#xff0c;便笺/录…

六、Json 数据的交互处理

文章目录 一、JSON 数据的交互处理1、为什么要使用 JSON2、JSON 和 JavaScript 之间的关系3、前端操作 JSON3.1 JavaScript 对象与 JSON 字符串之间的相互转换 4、JAVA 操作 JSON4.1 Json 的解析工具&#xff08;Gson、FastJson、Jackson&#xff09;4.2 ResponseBody 注解、Re…

2. HBase中文学习手册之如何运行一个单机版的HBase?

HBase中文学习手册之如何运行一个单机版的HBase? 1.1 介绍1.2 快速开始1.2.1 安装 Open JDK 81.2.2 启动 HBase1.2.3 Shell 练习1.2.4 运行停止脚本来停止HBase 1.1 介绍 上篇博文HBase中文学习手册之揭开Hbase的神秘面纱分享了 HBase 的一些理论基础知识的介绍。 本文将会继…

Spring boot(一)

Spring Boot是一个构建在Spring框架顶部的项目。它提供了一种简便&#xff0c;快捷的方式来设置&#xff0c;配置和运行基于Web的简单应用程序。 它是一个Spring模块&#xff0c;提供了 RAD(快速应用程序开发)功能。它用于创建独立的基于Spring的应用程序&#xff0c;因为它需…

mysql 字符集、比较规则, 比较规则底层逻辑

字符集的级别 show variables like ‘%charecter%’&#xff1b; character_set_server 服务器级别 一般在 5.7&#xff1a; C:\ProgramData\MySQL\MySQL Server 5.7\my.ini 8.0&#xff1a; C:\ProgramData\MySQL\MySQL Server 5.7\my.ini Linux 系列 vim /etc/my.cnf chara…

深入探讨C存储类和存储期——Storage Duration

&#x1f517; 《C语言趣味教程》&#x1f448; 猛戳订阅&#xff01;&#xff01;&#xff01; ​—— 热门专栏《维生素C语言》的重制版 —— &#x1f4ad; 写在前面&#xff1a;这是一套 C 语言趣味教学专栏&#xff0c;目前正在火热连载中&#xff0c;欢迎猛戳订阅&#…

MySQL—MySQL的NULL值是怎么存放的

一、引言 1、MySQL数据存放在哪个文件&#xff1f; 创建一个数据库会产生三种格式的文件&#xff0c;分别是.opt格式、.frm格式、.ibd格式。 opt格式&#xff1a;用来存储当前数据库的默认字符集和字符校验规则。 frm格式&#xff1a;该文件是用来保存每个表的元数据信息的&…

0基础入门C++之类和对象下篇

目录 1.再谈构造函数1.1构造函数赋值1.2初始化列表1.3explicit关键字 2.static成员2.1概念2.1静态成员变量2.2静态成员函数2.3特性 3.匿名对象4.友元函数4.1友元函数4.2友元类 5.内部类6.再次理解类和对象 1.再谈构造函数 首先我们先来回忆一下构造函数&#xff1a; 构造函数是…

【Java 高阶】一文精通 Spring MVC - 数据验证(七)

&#x1f449;博主介绍&#xff1a; 博主从事应用安全和大数据领域&#xff0c;有8年研发经验&#xff0c;5年面试官经验&#xff0c;Java技术专家&#xff0c;WEB架构师&#xff0c;阿里云专家博主&#xff0c;华为云云享专家&#xff0c;51CTO 专家博主 ⛪️ 个人社区&#x…

用AI + Milvus Cloud搭建着装搭配推荐系统

在上一篇文章中,我们学习了如何利用人工智能技术(例如开源 AI 向量数据库 Milvus Cloud 和 Hugging Face 模型)寻找与自己穿搭风格相似的明星。在这篇文章中,我们将进一步介绍如何通过对上篇文章中的项目代码稍作修改,获得更详细和准确的结果,文末附赠彩蛋。 注:试用此…

【头歌】构建哈夫曼树及编码

构建哈夫曼树及编码 第1关:构建哈夫曼树 任务描述 本关任务:构建哈夫曼树,从键盘读入字符个数n及这n个字符出现的频率即权值,构造带权路径最短的最优二叉树(哈夫曼树)。 相关知识 哈夫曼树的定义 设二叉树具有n个带权值的叶子结点{w1,w2,...,wn},从根结点到每个叶…

《Zookeeper》源码分析(二十三)之 客户端的命令处理过程

目录 客户端的命令处理过程1. ZooKeeper.create()2. ClientCnxn.submitRequest()3. SendThread.run()4. ClientCnxnSocket.doTransport()5. SendThread.readResponse() 客户端的命令处理过程 以创建节点命令为例&#xff0c;整个过程流程如下&#xff1a; CliCommand命令在抽…

uniapp国际化npm install vue-i18n报错

npm install vue-i18n //npmyarn add vue-i18n //yarn在vue2环境下&#xff0c;默认安装 npm install vue-i18n 的版本是 vue-i18n9.1.9&#xff0c;所以报错。 npm view vue-i18n versions --json 用以上命令查看版本&#xff1a; vue2建议5.0版本 npm install vue-i1…

Wireshark数据抓包分析之ARP协议

一、实验目的&#xff1a; 通过wireshark的数据抓包了解这个ARP协议的具体内容 二、预备知识: 1.Address Resolution Protocol协议&#xff0c;就是通过目标IP的值&#xff0c;获取到目标的mac地址的一个协议 2.ARP协议的详细工作过程&#xff0c;下面描述得非常清晰&#xff…