前往小程序，Get更优阅读体验！
立即前往
文档建议反馈控制台
首页
学习
活动
专区
工具
TVP
最新优惠活动
文章/答案/技术大牛
发布
首页
学习
活动
专区
工具
TVP最新优惠活动
返回腾讯云官网
社区首页 >专栏 >【实操记录】Oracle数据整库同步至Apache Doris

【实操记录】Oracle数据整库同步至Apache Doris

作者头像
程裕强
发布于 2023-10-18 17:34:47
5580
发布于 2023-10-18 17:34:47
举报
文章被收录于专栏：大数据学习笔记大数据学习笔记

本文是Oracle数据整库同步至Apache Doris实操记录，仅供参考

参考：https://cn.selectdb.com/blog/104

1、Oracle 配置
代码语言：javascript
复制
[root@node1 oracle]# pwd
/u01/app/oracle
[root@node1 oracle]# mkdir recovery_area
[root@node1 oracle]# chown -R oracle:dba recovery_area
代码语言：javascript
复制
SQL> conn sys/ as sysdba
输入口令: 
已连接。
SQL> alter system set db_recovery_file_dest_size = 2G;
系统已更改。
SQL> alter system set db_recovery_file_dest='/u01/app/oracle/recovery_area' scope=spfile;
系统已更改。
SQL>
在这里插入图片描述
在这里插入图片描述
代码语言：javascript
复制
SQL> shutdown immediate;
数据库已经关闭。
已经卸载数据库。
ORACLE 例程已经关闭。
SQL> startup mount;
ORACLE 例程已经启动。

Total System Global Area 1068937216 bytes
Fixed Size                  2233344 bytes
Variable Size             624954368 bytes
Database Buffers          436207616 bytes
Redo Buffers                5541888 bytes
数据库装载完毕。
SQL> alter database archivelog;

数据库已更改。

SQL> alter database open;

数据库已更改。

SQL> archive log list;
数据库日志模式            存档模式
自动存档             启用
存档终点            USE_DB_RECOVERY_FILE_DEST
最早的联机日志序列     1
下一个存档日志序列   2
当前日志序列           2
SQL> ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

数据库已更改。

SQL>
2、Oracle 数据
代码语言：javascript
复制
SQL> CREATE USER admin IDENTIFIED BY 123;

用户已创建。

SQL> GRANT dba TO admin;

授权成功。

SQL> quit;
代码语言：javascript
复制
[root@node1 ~]# sqlplus admin/123

SQL*Plus: Release 11.2.0.2.0 Production on 星期三 6月 28 08:06:29 2023

Copyright (c) 1982, 2011, Oracle.  All rights reserved.

连接到: 
Oracle Database 11g Express Edition Release 11.2.0.2.0 - 64bit Production
SQL>
代码语言：javascript
复制
CREATE TABLE t_person(ID NUMBER(10),NAME VARCHAR2(128) ,PRIMARY KEY(ID));
INSERT INTO t_person VALUES (1, 'zhangsan');
INSERT INTO t_person VALUES (2, 'lisi');
INSERT INTO t_person VALUES (3, 'wangwu');
CREATE TABLE t_dept(ID NUMBER(10),NAME VARCHAR2(128) ,PRIMARY KEY(ID));
INSERT INTO t_dept VALUES (1, 'HR');
INSERT INTO t_dept VALUES (2, 'IT');
commit;
3、启动Doris
代码语言：javascript
复制
[root@node1 doris-2.0]# fe/bin/start_fe.sh --daemon
[root@node1 doris-2.0]# be/bin/start_be.sh --daemon
[root@node1 doris-2.0]# jps
16081 DorisBE
14723 DorisFE
16158 Jps
[root@node1 doris-2.0]#
4、启动Flink

下载相关jar，复制到flink的lib目录

代码语言：javascript
复制
[root@node1 flink-1.17.1]# ll lib
总用量 270772
-rw-r--r-- 1  501 games    196491 5月  19 06:56 flink-cep-1.17.1.jar
-rw-r--r-- 1  501 games    542620 5月  19 06:59 flink-connector-files-1.17.1.jar
-rw-r--r-- 1  501 games    102472 5月  19 07:02 flink-csv-1.17.1.jar
-rw-r--r-- 1  501 games 135975541 5月  19 07:13 flink-dist-1.17.1.jar
-rw-r--r-- 1 root root    8452171 6月  27 23:12 flink-doris-connector-1.17-1.4.0.jar
-rw-r--r-- 1  501 games    180248 5月  19 07:02 flink-json-1.17.1.jar
-rw-r--r-- 1  501 games  21043319 5月  19 07:12 flink-scala_2.12-1.17.1.jar
-rw-r--r-- 1 root root   28702667 6月  28 08:28 flink-sql-connector-oracle-cdc-2.4.1.jar
-rw-r--r-- 1  501 games  15407424 5月  19 07:13 flink-table-api-java-uber-1.17.1.jar
-rw-r--r-- 1  501 games  38191226 5月  19 07:08 flink-table-planner-loader-1.17.1.jar
-rw-r--r-- 1  501 games   3146210 5月  19 06:56 flink-table-runtime-1.17.1.jar
-rw-r--r-- 1  501 games    208006 5月  17 06:07 log4j-1.2-api-2.17.1.jar
-rw-r--r-- 1  501 games    301872 5月  17 06:07 log4j-api-2.17.1.jar
-rw-r--r-- 1  501 games   1790452 5月  17 06:07 log4j-core-2.17.1.jar
-rw-r--r-- 1  501 games     24279 5月  17 06:07 log4j-slf4j-impl-2.17.1.jar
[root@node1 flink-1.17.1]#

启动flink集群

代码语言：javascript
复制
[root@node1 flink-1.17.1]# bin/start-cluster.sh 
Starting cluster.
Starting standalonesession daemon on host node1.
Starting taskexecutor daemon on host node1.
[root@node1 flink-1.17.1]# jps
11755 StandaloneSessionClusterEntrypoint
12091 TaskManagerRunner
12190 Jps
[root@node1 flink-1.17.1]#
5、测试同步操作
代码语言：javascript
复制
[root@node1 flink-1.17.1]# bin/flink run \
>     -Dexecution.checkpointing.interval=10s \
>     -Dparallelism.default=1 \
>     -c org.apache.doris.flink.tools.cdc.CdcTools \
>     lib/flink-doris-connector-1.17-1.4.0.jar \
>     oracle-sync-database \
>     --database test \
>     --oracle-conf hostname=node1 \
>     --oracle-conf port=1521 \
>     --oracle-conf username=admin \
>     --oracle-conf password=admin123 \
>     --oracle-conf database-name=XE \
>     --oracle-conf schema-name=ADMIN \
>     --including-tables "t_.*" \
>     --sink-conf fenodes=node1:8030 \
>     --sink-conf username=root \
>     --sink-conf password=123456\
>     --sink-conf jdbc-url=jdbc:mysql://node1:9030 \
>     --sink-conf sink.label-prefix=label \
>     --table-conf replication_num=1

Unknown operation oracle-sync-database
[root@node1 flink-1.17.1]#

异常：Unknown operation oracle-sync-database 处理办法： 需要使用最新的 flink-doris-connector 包https://repository.apache.org/content/repositories/snapshots/org/apache/doris/

代码语言：javascript
复制
[root@node1 flink-1.17.1]# ll lib
-rw-r--r-- 1  501 games    196491 5月  19 06:56 flink-cep-1.17.1.jar
-rw-r--r-- 1  501 games    542620 5月  19 06:59 flink-connector-files-1.17.1.jar
-rw-r--r-- 1  501 games    102472 5月  19 07:02 flink-csv-1.17.1.jar
-rw-r--r-- 1  501 games 135975541 5月  19 07:13 flink-dist-1.17.1.jar
-rw-r--r-- 1 root root    8526699 9月  21 12:46 flink-doris-connector-1.17-1.5.0-20230915.034854-2.jar
-rw-r--r-- 1  501 games    180248 5月  19 07:02 flink-json-1.17.1.jar
-rw-r--r-- 1  501 games  21043319 5月  19 07:12 flink-scala_2.12-1.17.1.jar
-rw-r--r-- 1 root root   28702667 6月  28 08:28 flink-sql-connector-oracle-cdc-2.4.1.jar
-rw-r--r-- 1  501 games  15407424 5月  19 07:13 flink-table-api-java-uber-1.17.1.jar
-rw-r--r-- 1  501 games  38191226 5月  19 07:08 flink-table-planner-loader-1.17.1.jar
-rw-r--r-- 1  501 games   3146210 5月  19 06:56 flink-table-runtime-1.17.1.jar
-rw-r--r-- 1  501 games    208006 5月  17 06:07 log4j-1.2-api-2.17.1.jar
-rw-r--r-- 1  501 games    301872 5月  17 06:07 log4j-api-2.17.1.jar
-rw-r--r-- 1  501 games   1790452 5月  17 06:07 log4j-core-2.17.1.jar
-rw-r--r-- 1  501 games     24279 5月  17 06:07 log4j-slf4j-impl-2.17.1.jar

重新执行

代码语言：javascript
复制
[root@node1 flink-1.17.1]# bin/flink run \
>     -Dexecution.checkpointing.interval=10s \
>     -Dparallelism.default=1 \
>     -c org.apache.doris.flink.tools.cdc.CdcTools \
>     lib/flink-doris-connector-1.17-1.5.0-20230915.034854-2.jar \
>     oracle-sync-database \
>     --database test \
>     --oracle-conf hostname=node1 \
>     --oracle-conf port=1521 \
>     --oracle-conf username=admin \
>     --oracle-conf password=admin123 \
>     --oracle-conf database-name=XE \
>     --oracle-conf schema-name=ADMIN \
>     --including-tables "t_.*" \
>     --sink-conf fenodes=node1:8030 \
>     --sink-conf username=root \
>     --sink-conf password=123456\
>     --sink-conf jdbc-url=jdbc:mysql://node1:9030 \
>     --sink-conf sink.label-prefix=label \
>     --table-conf replication_num=1


------------------------------------------------------------
 The program finished with the following exception:

org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: No tables to be synchronized.
        at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372)
        at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
        at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:105)
        at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:851)
        at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:245)
        at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1095)
        at org.apache.flink.client.cli.CliFrontend.lambda$mainInternal$9(CliFrontend.java:1189)
        at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
        at org.apache.flink.client.cli.CliFrontend.mainInternal(CliFrontend.java:1189)
        at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1157)
Caused by: java.lang.IllegalStateException: No tables to be synchronized.
        at org.apache.flink.util.Preconditions.checkState(Preconditions.java:193)
        at org.apache.doris.flink.tools.cdc.DatabaseSync.build(DatabaseSync.java:101)
        at org.apache.doris.flink.tools.cdc.CdcTools.syncDatabase(CdcTools.java:116)
        at org.apache.doris.flink.tools.cdc.CdcTools.createOracleSyncDatabase(CdcTools.java:80)
        at org.apache.doris.flink.tools.cdc.CdcTools.main(CdcTools.java:53)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
        ... 9 more
[root@node1 flink-1.17.1]#

报错：No tables to be synchronized 修改：--including-tables "T_.*"

代码语言：javascript
复制
[root@node1 flink-1.17.1]# bin/flink run \
>     -Dexecution.checkpointing.interval=10s \
>     -Dparallelism.default=1 \
>     -c org.apache.doris.flink.tools.cdc.CdcTools \
>     lib/flink-doris-connector-1.17-1.5.0-20230915.034854-2.jar \
>     oracle-sync-database \
>     --database test \
>     --oracle-conf hostname=node1 \
>     --oracle-conf port=1521 \
>     --oracle-conf username=admin \
>     --oracle-conf password=admin123 \
>     --oracle-conf database-name=XE \
>     --oracle-conf schema-name=ADMIN \
>     --including-tables "T_.*" \
>     --sink-conf fenodes=node1:8030 \
>     --sink-conf username=root \
>     --sink-conf password=123456\
>     --sink-conf jdbc-url=jdbc:mysql://node1:9030 \
>     --sink-conf sink.label-prefix=label \
>     --table-conf replication_num=1


------------------------------------------------------------
 The program finished with the following exception:

org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: The following SQL query could not be executed: SELECT `SCHEMA_NAME` FROM `INFORMATION_SCHEMA`.`SCHEMATA`;
        at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372)
        at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
        at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:105)
        at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:851)
        at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:245)
        at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1095)
        at org.apache.flink.client.cli.CliFrontend.lambda$mainInternal$9(CliFrontend.java:1189)
        at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
        at org.apache.flink.client.cli.CliFrontend.mainInternal(CliFrontend.java:1189)
        at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1157)
Caused by: org.apache.doris.flink.exception.DorisSystemException: The following SQL query could not be executed: SELECT `SCHEMA_NAME` FROM `INFORMATION_SCHEMA`.`SCHEMATA`;
        at org.apache.doris.flink.catalog.doris.DorisSystem.extractColumnValuesBySQL(DorisSystem.java:137)
        at org.apache.doris.flink.catalog.doris.DorisSystem.listDatabases(DorisSystem.java:59)
        at org.apache.doris.flink.catalog.doris.DorisSystem.databaseExists(DorisSystem.java:67)
        at org.apache.doris.flink.tools.cdc.DatabaseSync.build(DatabaseSync.java:102)
        at org.apache.doris.flink.tools.cdc.CdcTools.syncDatabase(CdcTools.java:116)
        at org.apache.doris.flink.tools.cdc.CdcTools.createOracleSyncDatabase(CdcTools.java:80)
        at org.apache.doris.flink.tools.cdc.CdcTools.main(CdcTools.java:53)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
        ... 9 more
Caused by: java.lang.ClassNotFoundException: com.mysql.jdbc.Driver
        at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
        at org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:67)
        at org.apache.flink.util.ChildFirstClassLoader.loadClassWithoutExceptionHandling(ChildFirstClassLoader.java:74)
        at org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:51)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
        at java.lang.Class.forName0(Native Method)
        at java.lang.Class.forName(Class.java:264)
        at org.apache.doris.flink.connection.SimpleJdbcConnectionProvider.getOrEstablishConnection(SimpleJdbcConnectionProvider.java:52)
        at org.apache.doris.flink.catalog.doris.DorisSystem.extractColumnValuesBySQL(DorisSystem.java:121)
        ... 20 more
[root@node1 flink-1.17.1]#

缺少mysql驱动，flink的lib目录添加mysql-connector-java-5.1.49.jar

代码语言：javascript
复制
[root@node1 flink-1.17.1]# bin/flink run \
>     -Dexecution.checkpointing.interval=10s \
>     -Dparallelism.default=1 \
>     -c org.apache.doris.flink.tools.cdc.CdcTools \
>     lib/flink-doris-connector-1.17-1.5.0-20230915.034854-2.jar \
>     oracle-sync-database \
>     --database test \
>     --oracle-conf hostname=node1 \
>     --oracle-conf port=1521 \
>     --oracle-conf username=admin \
>     --oracle-conf password=admin123 \
>     --oracle-conf database-name=XE \
>     --oracle-conf schema-name=ADMIN \
>     --including-tables "T_.*" \
>     --sink-conf fenodes=node1:8030 \
>     --sink-conf username=root \
>     --sink-conf password=123456\
>     --sink-conf jdbc-url=jdbc:mysql://node1:9030 \
>     --sink-conf sink.label-prefix=label \
>     --table-conf replication_num=1

Thu Sep 21 13:06:59 EDT 2023 WARN: Establishing SSL connection without server's identity verification is not recommended. According to MySQL 5.5.45+, 5.6.26+ and 5.7.6+ requirements SSL connection must be established by default if explicit option isn't set. For compliance with existing applications not using SSL the verifyServerCertificate property is set to 'false'. You need either to explicitly disable SSL by setting useSSL=false, or set useSSL=true and provide truststore for server certificate verification.
Job has been submitted with JobID fd244bcb10f06e5aa801d2708441bb86

登录Doris查看同步效果

代码语言：javascript
复制
[root@node1 doris-2.0]# mysql -h192.168.203.101 -P9030 -uroot -p
Enter password: 
Welcome to the MySQL monitor.  Commands end with ; or \g.
Your MySQL connection id is 3
Server version: 5.7.99 Doris version doris-2.0.1.1-ea0bfb2

Copyright (c) 2000, 2021, Oracle and/or its affiliates. All rights reserved.

Oracle is a registered trademark of Oracle Corporation and/or its
affiliates. Other names may be trademarks of their respective
owners.

Type 'help;' or '\h' for help. Type '\c' to clear the current input statement.

mysql> use test;
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 |
+----------------+
| T_DEPT         |
| T_PERSON       |
+----------------+
2 rows in set (0.00 sec)

mysql> select * from T_PERSON;
Empty set (0.29 sec)

mysql> select * from T_DEPT;
Empty set (0.06 sec)
6、同步成功
代码语言：javascript
复制
bin/flink run \
    -Dexecution.checkpointing.interval=10s \
    -Dparallelism.default=1 \
    -c org.apache.doris.flink.tools.cdc.CdcTools \
    lib/flink-doris-connector-1.17-1.5.0-20230915.034854-2.jar \
    oracle-sync-database \
    --database test \
    --oracle-conf hostname=node1 \
    --oracle-conf port=1521 \
    --oracle-conf username=admin \
    --oracle-conf password=123 \
    --oracle-conf database-name=XE \
    --oracle-conf schema-name=ADMIN \
    --including-tables "T_.*" \
    --sink-conf fenodes=node1:8030 \
    --sink-conf username=root \
    --sink-conf password=123456\
    --sink-conf jdbc-url=jdbc:mysql://node1:9030 \
    --sink-conf sink.label-prefix=label \
    --table-conf replication_num=1
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

查看Flink中运行的Job http://192.168.203.101:8081/

在这里插入图片描述
在这里插入图片描述
7、测试新增记录同步
代码语言：javascript
复制
SQL> INSERT INTO t_person VALUES (4, 'test_user');

已创建 1 行。

SQL> INSERT INTO t_dept VALUES (3, 'CS');

已创建 1 行。

SQL> commit;

提交完成。

SQL>
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
本文参与 腾讯云自媒体分享计划，分享自作者个人站点/博客。
原始发表：2023-10-05，如有侵权请联系 cloudcommunity@tencent.com 删除
oracle
apache
flink
数据
同步

本文分享自 作者个人站点/博客 前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体分享计划  ，欢迎热爱写作的你一起参与！

oracle
apache
flink
数据
同步
评论
登录后参与评论
0 条评论
热度
最新
登录 后参与评论
推荐阅读
LV.
文章
0
获赞
0
目录
  • 1、Oracle 配置
  • 2、Oracle 数据
  • 3、启动Doris
  • 4、启动Flink
  • 5、测试同步操作
  • 6、同步成功
  • 7、测试新增记录同步
相关产品与服务
云数据库 MySQL
腾讯云数据库 MySQL（TencentDB for MySQL）为用户提供安全可靠，性能卓越、易于维护的企业级云数据库服务。其具备6大企业级特性，包括企业级定制内核、企业级高可用、企业级高可靠、企业级安全、企业级扩展以及企业级智能运维。通过使用腾讯云数据库 MySQL，可实现分钟级别的数据库部署、弹性扩展以及全自动化的运维管理，不仅经济实惠，而且稳定可靠，易于运维。
免费体验产品介绍产品文档
云数据库采购特惠，高性能、高可靠、高安全，超值优惠等你享！
领券

腾讯云开发者

扫码关注腾讯云开发者

扫码关注腾讯云开发者

领取腾讯云代金券

热门产品

热门推荐

更多推荐

Copyright © 2013 - 2024 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有 

深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 深公网安备号 44030502008569

腾讯云计算（北京）有限责任公司 京ICP证150476号 |  京ICP备11018762号 | 京公网安备号11010802020287

问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档

Copyright © 2013 - 2024 Tencent Cloud.

All Rights Reserved. 腾讯云 版权所有

登录 后参与评论