江苏龙网

  1. 主页 > 生活百科 > >

基于OGG 实现Oracle到Kafka增量数据实时同步

Oracle

背景在大数据时代，存在大量基于数据的业务 。数据需要在不同的系统之间流动、整合 。通常，核心业务系统的数据存在OLTP数据库系统中，其它业务系统需要获取OLTP系统中的数据 。传统的数仓通过批量数据同步的方式，定期从OLTP系统中抽取数据 。但是随着业务需求的升级，批量同步无论从实时性，还是对在线OLTP系统的抽取压力，都无法满足要求 。需要实时从OLTP系统中获取数据变更，实时同步到下游业务系统 。
本文基于Oracle OGG，介绍一种将Oracle数据库的数据实时同步到Kafka消息队列的方法 。
Kafka是一种高效的消息队列实现，通过订阅kafka的消息队列，下游系统可以实时获取在线Oracle系统的数据变更情况，实现业务系统 。
环境介绍组件版本

基于OGG 实现Oracle到Kafka增量数据实时同步

文章插图
 
整体架构图
基于OGG 实现Oracle到Kafka增量数据实时同步

文章插图
 
名词解释
1.OGG Manager
OGG Manager用于配置和管理其它OGG组件，配置数据抽取、数据推送、数据复制，启动和停止相关组件，查看相关组件的运行情况 。
2.数据抽取（Extract）
抽取源端数据库的变更（DML, DDL） 。数据抽取主要分如下几种类型：
本地抽取
从本地数据库捕获增量变更数据，写入到本地Trail文件
数据推送（Data Pump）
从本地Trail文件读取数据，推送到目标端 。
初始数据抽取
从数据库表中导出全量数据，用于初次数据加载
3.数据推送（Data Pump）
Data Pump是一种特殊的数据抽取（Extract）类型，从本地Trail文件中读取数据，并通过网络将数据发送到目标端OGG
4.Trail文件
数据抽取从源端数据库抓取到的事物变更信息会写入到Trail文件 。
5.数据接收（Collector）
数据接收程序运行在目标端机器，用于接收Data Pump发送过来的Trail日志，并将数据写入到本地Trail文件 。
6.数据复制（Replicat）
数据复制运行在目标端机器，从Trail文件读取数据变更，并将变更数据应用到目标端数据存储系统 。本案例中，数据复制将数据推送到kafka消息队列 。
7.检查点（Checkpoint）
检查点用于记录数据库事物变更 。
 
操作步骤源端Oracle配置
1.检查归档
使用OGG，需要在源端开启归档日志
SQL> archive log list;     Database log mode              Archive Mode     Automatic archival             Enabled     Archive destination            /u01/App/oracle/product/12.2.0/db_1/dbs/arch     Oldest online log sequence     2576     Next log sequence to archive   2577     Current log sequence           25772.检查数据库配置
SQL> select force_logging, supplemental_log_data_min from v$database; FORCE_LOGG SUPPLEMENTAL_LOG_DATA_MI ---------- ------------------------ YES        YES如果没有开启辅助日志，需要开启：
SQL> alter database force logging; SQL> alter database add supplemental log data;3.开启goldengate复制参数
SQL> alter system set enable_goldengate_replication = true;4.创建源端Oracle账号
SQL> create tablespace tbs_ogg datafile '/oradata/dtstack/tbs_ogg.dbf' size 1024M autoextend on; SQL> create user ggsadmin identified by oracle default tablespace tbs_ogg; SQL> grant dba to ggsadmin;5.创建测试表
SQL> create table baiyang.ora_to_kfk as select OWNER, OBJECT_NAME, SUBOBJECT_NAME, OBJECT_ID, DATA_OBJECT_ID, OBJECT_TYPE from all_objects where object_id < 500; SQL> alter table baiyang.ora_to_kfk add constraint pk_kfk_obj primary key(object_id); SQL> select count(*) from baiyang.ora_to_kfk;    COUNT(*) ----------         436源端OGG配置
1.检查源端OGG环境
cd /oradata/oggorcl/ogg ./ggsci GGSCI (dtproxy) 1> info all Program     Status      Group       Lag at Chkpt  Time Since ChkptMANAGER     STOPPED2.创建相关文件夹
GGSCI (dtproxy) 2> create subdirs     Creating subdirectories under current directory /oradata/oggorcl/ogg    Parameter file                 /oradata/oggorcl/ogg/dirprm: created.     Report file                    /oradata/oggorcl/ogg/dirrpt: created.     Checkpoint file                /oradata/oggorcl/ogg/dirchk: created.     Process status files           /oradata/oggorcl/ogg/dirpcs: created.     SQL script files               /oradata/oggorcl/ogg/dirsql: created.     Database definitions files     /oradata/oggorcl/ogg/dirdef: created.     Extract data files             /oradata/oggorcl/ogg/dirdat: created.     Temporary files                /oradata/oggorcl/ogg/dirtmp: created.     Credential store files         /oradata/oggorcl/ogg/dircrd: created.     Masterkey wallet files         /oradata/oggorcl/ogg/dirwlt: created.     Dump files                     /oradata/oggorcl/ogg/dirdmp: created


推荐阅读


上一篇：Python爬虫遇到验证码的几种处理方式，文章末尾有源码

下一篇：聊聊HBase分布式数据库设计那些事