如何使用StreamSets实现Oracle中变化数据实时写入Kudu

温馨提示:如果使用电脑查看图片不清晰,可以使用手机打开文章单击文中的图片放大查看高清原图。

Fayson的github: https://github.com/fayson/cdhproject

提示:代码块部分可以左右滑动查看噢

1.文档编写目的


在前面的文章Fayson介绍了关于StreamSets的一些文章,参考《如何在CDH中安装和使用StreamSets》、《如何使用StreamSets从MySQL增量更新数据到Hive》、《如何使用StreamSets实现MySQL中变化数据实时写入Kudu》、《如何使用StreamSets实时采集Kafka并入库Kudu》、《如何使用StreamSets实现MySQL中变化数据实时写入HBase》、《如何使用StreamSets实时采集Kafka数据并写入Hive表》和《如何使用StreamSets实时采集Kafka中嵌套JSON数据并写入Hive表》,本篇文章Fayson主要介绍如何使用StreamSets实时采集Oracle中的变化数据实时写入Kudu,StreamSets的流程处理如下:

  • 内容概述

1.配置StreamSets创建Pipeline及测试

2.总结

3.附录:脚本代码

  • 测试环境

1.CentOS7.2

2.CM和CDH版本为cdh5.14

3.Oracle11.2.0.4

4.StreamSets3.2.0

  • 前置条件

1.集群已经安装StreamSets并运行正常

2.Oracle及Oracle LogMiner运行正常

2.配置StreamSets创建Pipeline及测试


1.在CDH集群中cdh04节点添加StreamSets DataCollector实例服务并启动。

2.在浏览器中输入http://cdh04:18630/,即可访问StreamStes Web Console服务。针对此测试场景新建一个命名为oracle_sdc_kudu的Pipeline,保存后开始构建对ORACLE Database实时数据同步的数据流管道。

3.在 Origin Processer中选择 Oracle CDC Client。

4.在 Destination Processor中选择 Kudu-Apache Kudu 1.6.0。此CDH测试集群中已部署Kudu 1.6.0 版本。

5.根据测试场景需求描述,在StreamSets中将命名为oracle_sdc_kudu 的Pipeline配置为Oracle CDC Client 与 Kudu 的端到端实时数据同步模式。

6. 将StreamSets Origin 端 Oracle CDC Client 服务的关键配置按需调整如下。

  • Tables 配置为ORACLE DATABASE中的定义的测试表TEST.SDC,注意ORACEL大小写敏感。
  • Initial Change配置为 From Latest Change,实时数据同步只需要捕获变化数据。
  • Operation配置为KUDU对应支持的CRUD操作类型,匹配ORACLE Database中实时变化数据的增删改操作事务操作。
  • JDBC Connection String 配置为源端Oracle Database 服务的JDBC访问连接串:jdbc:oracle:thin:@10.89.180.21:1521:orcl。
  • USE CREDENTIALS 配置为开启,便于配置ORACLE DATABASE访问认证信息。
  • JDBC Driver Class Name 配置为ORACLE DATABASE R2 11.2.0.4提供的ojdbc6.jar 中定义的JDBC驱动程序fullPath: oracle.jdbc.driver.OracleDriver。

7.将StreamSets Destination 端 Kudu 服务的关键配置按需调整如下

  • Stage Library配置为CDH 5.14.0安装包中内置的对应KUDU版本。
  • Kudu Masters 配置为 CDH集群中Kudu Master服务地址:10.89.180.11:7051。
  • Table Name 配置为实时数据同步到Kudu中目标表:impala::POC_ENDING.sdc。
  • Field Column Mapping 配置为 SDC与Kudu的字段映射关系。
  • Default Operation 配置为INSERT类型的SDC缺省操作事件,以匹配Kudu的KV数据库模式,基于主键的自动更新Streamsets中实时数据同步产生的ORACLE Database Insert and update事件。
  • Change Format Log 配置为此测试场景中StreamSets Origin端定义的Oracle CDC Client类型,以解析实时数据同步ORCLE DATABASE的REDO LOG事件与操作。

8.添加Oracle Database 的JDBC驱动程序包ojdbc.jar。

  • 在Cloudera Manager中对StreamSets服务管理其资源服务目录信息及权限。

详情可参考:https://streamsets.com/documentation/datacollector/latest/help。

9.针对此测试场景,完成以上配置步骤后,执行validate操作,验证无误即可运行此Pipeline,开始ORACLE to KUDU的实时数据同步。

StreamSets Pipeline开启后,正常运行状态及其监控图表与日志查看。

10.验证新增数据实时同步。

  • 在Oracle Database sqlplus中执行以下脚本内容。
insert into test.sdc(id,name) values('1','AAA');
insert into test.sdc(id,name) values('2','BBB');
insert into test.sdc(id,name) values('3','CCC');
insert into test.sdc(id,name) values('4','DDD');
insert into test.sdc(id,name) values('5','EEE');
insert into test.sdc(id,name) values('6','FFF');
commit;

(可左右滑动)

  • StreamSet 监控信息。
  • 在Impala-shell中执行以下脚本内容。
select * from poc_ending.sdc;
select count(1) from poc_ending.sdc;

(可左右滑动)

11.验证修改数据的实时同步。

  • 在Oracle Database sqlplus中执行以下脚本内容。
update test.sdc set name='111' where id ='1';
update test.sdc set name='222' where id ='2';
update test.sdc set name='333' where id ='3';
update test.sdc set name='444' where id ='4';
update test.sdc set name='555' where id ='5';
update test.sdc set name='666' where id ='6';
commit;

(可左右滑动)

  • StreamSet 监控信息。
  • 在Impala-shell中执行以下脚本内容。
select * from poc_ending.sdc;
select count(1) from poc_ending.sdc;

(可左右滑动)

12.验证删除数据的实时同步。

  • 在Oracle Database sqlplus中执行以下脚本内容。
delete from test.sdc where id ='1';
delete from test.sdc where id ='2';
delete from test.sdc where id ='3';
delete from test.sdc where id ='4';
delete from test.sdc where id ='5';
delete from test.sdc where id ='6';
commit;

(可左右滑动)

  • StreamSet 监控信息。
  • 在Impala-shell中执行以下脚本内容。
select count(1) from poc_ending.sdc;

(可左右滑动)

3.总结


1.满足企业从Oracle数据库到Hadoop(kudu)的实时数据同步功能需求。

2.Cloudera Manager支持对CDH服务组件与StreamSets服务组件的统一管理。

3.StreamSets支持从Oracle Database 11.2.0.4 到 Kudu 1.6.0的端到端实时数据同步,在Oracle LogMiner的支持下可以实现对Oracle Database数据库产生的REDO日志文件分析并将其INSERT/UPDATE/DELETE等影响数据变化的事件实时同步到KDUD中。

4.StreamSets 提供完善的Web Console管理服务,实现可视化数据同步配置及管理、运行过程监控、异常告警、事件日志查看等。

4.附录:脚本代码


-- PERFORM on IMPALA WITH KUDU --------------------------------------------------
CREATE DATABASE IF NOT EXISTS POC_ENDING;
CREATE TABLE IF NOT EXISTS POC_ENDING.sdc(
    id string,
    name string,
PRIMARY KEY (id)
) PARTITION BY HASH(id) PARTITIONS 12
STORED AS KUDU
TBLPROPERTIES('kudu.master_addresses'='cdh01:7051');

-- PERFORM on ORACLE DATABASE --------------------------------------------------
-- STREAMSETS CDC ORIGIN CLIENT for ORACLE DATABASE is dependeced ORACLE LogMiner session and view v$logmnr_contents.
-- 1.配置对象数据字典:通过ORACLE内置存储过程DBMS_LOGMNR_D.BUILD可配置当前联机数据字典/平面数据字典文件/重做日志字典文件。
-- 2.追加重做日志文件:通过ORACLE内置存储过程DBMS_LOGMNR.ADD_LOGFILE可配置指定历史文件/指定滚动文件。
-- 3.开启数据变更分析:通过ORACLE内置存储过程DBMS_LOGMNR.START_LOGMNR启动LogMiner后可查询分析视图v$logmnr_contents。
-- 4.停止数据变更分析:通过ORACLE内置存储过程DBMS_LOGMNR.END_LOGMNR可停止LogMiner的SESSION级访问。
SQL> select * from v$logmnr_contents;
EXCEPTION 1 ON ORACLE-SESSION-CLIENT:
ORA-01306: dbms_logmnr.start_logmnr() must be invoked before selecting from v$logmnr_contents
SQL> execute DBMS_LOGMNR.START_LOGMNR(OPTIONS=> DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG);
ORA-01292: no log file has been specified for the current LogMiner session
ORA-06512: at "SYS.DBMS_LOGMNR", line 58
ORA-06512: at line 1
SQL> execute DBMS_LOGMNR_D.BUILD(OPTIONS=> DBMS_LOGMNR_D.STORE_IN_REDO_LOGS);
SQL> execute DBMS_LOGMNR.START_LOGMNR(OPTIONS=> DBMS_LOGMNR.DICT_FROM_REDO_LOGS);
EXCEPTION 2 ON ORACLE-SESSION-CLIENT and STREAMSETS-WEBCONSOLE:
java.sql.SQLException: ORA-01292: no log file has been specified for the current LogMiner session
ORA-06512: at "SYS.DBMS_LOGMNR", line 58
ORA-06512: at line 1
[oracle@oracleasm ~]$ oerr ora 1292
01292, 00000, "no log file has been specified for the current LogMiner session"
// *Cause:  No logfile has been specified for the LogMiner session.
// *Action: Specify atleast one log file.
EXCEPTION 3 ON ORACLE-SESSION-CLIENT and STREAMSETS-WEBCONSOLE:
java.sql.SQLException: ORA-01291: Not all logfiles corresponding to the time or scn range specified
ORA-06512: 在 "SYS.DBMS_LOGMNR", line 58
ORA-06512: 在 line 1
[root@oracleasm dmp]# oerr ora 1291
01291, 00000, "missing logfile"
// *Cause:  Not all logfiles corresponding to the time or scn range specified
//          have been added to the list. 
// *Action: Check the v$logmnr_logs view to determine the missing scn 
//          range, and add the relevant logfiles.
SOLUTION:
SQL> select member from v$logfile;
/mnt/orcl/orcl/redo03.log
/mnt/orcl/orcl/redo02.log
/mnt/orcl/orcl/redo01.log
SQL> execute DBMS_LOGMNR.add_logfile('/mnt/orcl/orcl/redo01.log');
SQL> execute DBMS_LOGMNR.add_logfile('/mnt/orcl/orcl/redo02.log');
SQL> execute DBMS_LOGMNR.add_logfile('/mnt/orcl/orcl/redo03.log');
--------------------------------------------------

sqlplus / as sysdba;
SQL> show user;
--IF select log_mode from v$database that result is no archivelog;
SQL> shutdown immediate;
SQL> startup mount;
SQL> alter database archivelog;
SQL> alter database open;
--IF select supplemental_log_data_min,supplemental_log_data_pk,supplemental_log_data_all from V$database that result is no;
SQL> alter database add supplemental log data;
SQL> alter database add supplemental log data (primary key) columns;
SQL> alter database add supplemental log data (all) columns;
SQL> alter system switch logfile;
SQL> select log_mode,supplemental_log_data_min,supplemental_log_data_pk,supplemental_log_data_all from V$database;
LOG_MODE                 SUPPLEMENTAL_LOG SUPPLE SUPPLE
------------------------ ---------------- ------ ------
ARCHIVELOG               YES              YES    YES

SQL> --show parameter utl_file_dir;
SQL> --show parameter log_archived_dest_1;
SQL> --select * from v$log;
SQL> --select * from v$logfile;
SQL> --create table test.sdc (id varchar2(20), name varchar2(20));
SQL> create user test identified by test;
SQL> grant create session, alter session, execute_catalog_role,select any dictionary, select any transaction, select any table to test;
SQL> grant select on V$logmnr_parameters to test;
SQL> grant select on v$logmnr_logs to test;
SQL> grant select on v$archived_log to test;
SQL> grant select on test.sdc to test;

SQL> conn test/test;
SQL> show user;
SQL> desc v$logmnr_contents
SQL> 
SQL> --execute DBMS_LOGMNR.END_LOGMNR;
SQL> execute DBMS_LOGMNR.add_logfile('/mnt/orcl/orcl/redo01.log');
SQL> execute DBMS_LOGMNR.add_logfile('/mnt/orcl/orcl/redo02.log');
SQL> execute DBMS_LOGMNR.add_logfile('/mnt/orcl/orcl/redo03.log');
SQL> execute DBMS_LOGMNR.START_LOGMNR(OPTIONS=> DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG);
SQL> SELECT SCN,TIMESTAMP,SEQUENCE#,SQL_REDO,SQL_UNDO from v$logmnr_contents WHERE TABLE_NAME='SDC';


SQL> insert into test.sdc(id,name) values('1','AAA');
SQL> insert into test.sdc(id,name) values('2','BBB');
SQL> insert into test.sdc(id,name) values('3','CCC');
SQL> insert into test.sdc(id,name) values('4','DDD');
SQL> insert into test.sdc(id,name) values('5','EEE');
SQL> insert into test.sdc(id,name) values('6','FFF');
SQL> commit;
SQL> SELECT SCN,TIMESTAMP,SEQUENCE#,SQL_REDO,SQL_UNDO from v$logmnr_contents WHERE TABLE_NAME='SDC';
SQL> update test.sdc set name='111' where id ='1';
SQL> update test.sdc set name='222' where id ='2';
SQL> update test.sdc set name='333' where id ='3';
SQL> update test.sdc set name='444' where id ='4';
SQL> update test.sdc set name='555' where id ='5';
SQL> update test.sdc set name='666' where id ='6';
SQL> commit;
SQL> SELECT SCN,TIMESTAMP,SEQUENCE#,SQL_REDO,SQL_UNDO from v$logmnr_contents WHERE TABLE_NAME='SDC';
SQL> delete from test.sdc where id ='1';
SQL> delete from test.sdc where id ='2';
SQL> delete from test.sdc where id ='3';
SQL> delete from test.sdc where id ='4';
SQL> delete from test.sdc where id ='5';
SQL> delete from test.sdc where id ='6';
SQL> commit;
SQL> SELECT SCN,TIMESTAMP,SEQUENCE#,SQL_REDO,SQL_UNDO from v$logmnr_contents WHERE TABLE_NAME='SDC';

(可左右滑动)

提示:代码块部分可以左右滑动查看噢

为天地立心,为生民立命,为往圣继绝学,为万世开太平。 温馨提示:要看高清无码套图,请使用手机打开并单击图片放大查看。

推荐关注Hadoop实操,第一时间,分享更多Hadoop干货,欢迎转发和分享。

原创文章,欢迎转载,转载请注明:转载自微信公众号Hadoop实操

原文发布于微信公众号 - Hadoop实操(gh_c4c535955d0f)

原文发表时间:2018-07-06

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏杨建荣的学习笔记

使用shell生成orabbix自动化配置脚本(r6笔记第53天)

在使用Orabbix监控Oracle的时候,本身和zaabix agent最大的不同便是使用Orabbix不需要对每个数据库实例都安装单独的agent,而是一个...

33080
来自专栏乐沙弥的世界

基于同一主机配置Oracle 11g Data Guard(logical standby)

      Oracle Data Guard逻辑备库是利用主库的一个备份首先建立一个物理备库,然后再将其转换为逻辑备库。这之后主库将日志传递到备库,备库利用l...

10110
来自专栏乐沙弥的世界

启用用户进程跟踪

仅仅需要标识该会话并为该会话启用跟踪(专用模式为一对一模式,即一个用户进程对应一个服务器进程)

9920
来自专栏沃趣科技

Log Miner 挖挖挖

Log Miner是Oracle自Oracle 8i以后推出的一个可以分析数据库redo log和archivelog内容的工具,可以通过日志分析所有对数据库的...

16550
来自专栏乐沙弥的世界

只读表空间的备份与恢复

--====================== --  只读表空间的备份与恢复 --====================== 一、只读表空间的特性...

10220
来自专栏「3306 Pai」社区

浅析ProxySQL用户管理

对于读写分离特别重要,保证了同一个事务中所有的语句都会路由到同一组示例,防止出现同一个事务中,上下文数据不一致的情况。例如,在不开启这个属性的情况下:

32610
来自专栏更流畅、简洁的软件开发方式

检查两个数据库里的表名、字段是否一致的一种方法

不知道大家有没有遇到过这种情况。 程序已经给客户安装上了,并且客户已经录入了一些信息,然后程序还需要作比较大的变动(修改功能、增加模块等),数据库就不可避免要...

30080
来自专栏Hadoop实操

如何使用Sentry通过视图实现Impala的行级授权

温馨提示:要看高清无码套图,请使用手机打开并单击图片放大查看。 Fayson的github:https://github.com/fayson/cdhproje...

59190
来自专栏杨建荣的学习笔记

系统级alias vs Oracle ADR功能(r5笔记第35天)

Oracle在11g中推出的新特性ADR,即Automatic Diagnostic Repository 个人理解这个工具就是能够高效的把一些日志文件轻松管理...

37180
来自专栏CSDN技术头条

MySQL 5.7 X Plugin:流水线技术vs.并行查询技术

本文将以MySQL 5.7 X Plugin为例,对比分析流水线(pipelining)和并行查询技术。 另一篇博文《MySQL 5.7 X Plugin支持异...

43060

扫码关注云+社区

领取腾讯云代金券