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

Fayson

发布于 2018-07-12 16:01:22

5K0

发布于 2018-07-12 16:01:22

温馨提示：如果使用电脑查看图片不清晰，可以使用手机打开文章单击文中的图片放大查看高清原图。
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';

（可左右滑动）