Oracle Streams: Step by Step - Database Wisdom

1 Oracle Streams: step by step What is Oracle Streams? What is Advanced Queues (AQ)? What is Change Data Capture (CDC)? What is a Logical Change record (LCR)? How Streams Works How do you configure Streams, AQ and CDC? How do you use Streams in your programs? Adding records automatically Adding record programmatically Pulling records off the queueWho am I Lewis R Cunningham Oracle ACE Certified PL/SQL Developer Author Blogger Database Geek Member of SOUG, IOUG, ODTUG, ACMWhat is Oracle Streams? Streams enables data and event movement Movement may be within or between databases Can implement replication Data and events can be captured explicitly: programatically Implicitly: redo log Performant Log mining reduces overhead Downstreams mining reduces it even moreWhat is Oracle Streams?

2 , cont'd Streams Features Distinct capture/propagation/apply services Message Queueing (via AQ) Publish Subscribe Rules based Capture/Transform/Apply Database integration Easy configuration Flexible Inline transformationsWhat is Oracle AQ? Oracle 's messaging solution Queue based Persistent queues protect data Buffered, non-persistent queues offer maximum performance Oracle provides all of the maintenance code Secure Queue Level Access Control Messages Types Raw XML Object AnydataWhat is Oracle AQ?, cont'd Publish/Subscribe Multi-consumer Multi-message types RAC integration API support OCI (C), OCCI (C++) Java JMS PL/SQL OO40 (COM) Internet Data Access (IDAP) ODBC via Heterogeneous GatewaysWhat is Change Data Capture? CDC efficiently identifies changed data and make it available for downstrean use CDC enables efficient warehouse loading No flat files Direct data movement Changes are queued Multiple subscribers are allowed Subscribers can receive changed data Data may be transformed Changes can include DDL and DML LCR = Logical Change record Committed records generate an LCR to the redo log An LCR shows the difference between an existing record and the new record For DDL, the record is a Database object For DML, the record is table dataWhat is an LCR?

3 What is an LCR? 10 How it worksSource DatabaseTarget DatabaseTable11 Data ChangesSource DatabaseTarget DatabaseTableLogWriterData Manipulation12DB Writes to OS FileSource DatabaseTarget DatabaseTableLogWriterREDO LOGW rites To13 Streams Takes OverSource DatabaseTarget DatabaseTableLogWriterREDO LOGS treamsCaptureCapture Process Reads Redo and Extracts the Logical Change record (LCR)14 Streams Queues The LCRS ource DatabaseTarget DatabaseTableLogWriterREDO LOGS treamsCaptureStreams QueueStreams Capture Process writes the LCR to a local OR REMOTE Streams Queue Table15 LCR PropagatesSource DatabaseTarget DatabaseTableLogWriterREDO LOGS treamsCaptureStreams QueueStreams Capture Process writes the LCR to a local OR REMOTE Streams Queue TableStreams Queue16 Changes Are AppliedSource DatabaseTarget DatabaseTableLogWriterREDO LOGS treamsCaptureStreams QueueStreams Capture Process writes the LCR to a local OR REMOTE Streams Queue TableStreams QueueStreamsApplyStreams Apply can also transform for one record to many, add columns.

4 Streams Overview - US IFST arget DatabaseSource DatabaseTarget DatabaseTableLogWriterREDO LOGS treamsCaptureStreams QueueStreams Capture Process writes the LCR to a local OR REMOTE Streams Queue TableStreams QueueStreamsApplyTableTableTableConfigur ing Streams The Scenario Capture changes to the Employee table in 10g ORCL instance) Send changes to Employee_audit in 9i (SECOND instance) We will start with capture and apply in 10g Once the capture and apply is working, we will modify the stream to send the data to 9iConfiguring Streams We need to create the Admin user and streams tablespace sqlplus / as sysdba create tablespace streams_tbs datafile 'c:\temp\ ' size 25M reuse autoextend on maxsize unlimited; create user strmadmin identified by strmadmindefault tablespace streams_tbsquota unlimited on streams_tbs;Configuring Streams We need to grant the appropriate permissions grant dba to strmadmin; BEGIN ( grantee => 'strmadmin', grant_privileges => true);END;/ conn hr/hr grant all on to strmadmin.

5 Configuring Streams Create the Audit Table CREATE TABLE employee_audit( employee_id NUMBER(6), first_name VARCHAR2(20), last_name VARCHAR2(25), email VARCHAR2(25), phone_number VARCHAR2(20), hire_date DATE, job_id VARCHAR2(10), salary NUMBER(8,2), commission_pct NUMBER(2,2), manager_id NUMBER(6), department_id NUMBER(4), upd_date DATE, user_name VARCHAR2(30), action VARCHAR2(30));Configuring Streams Grant read on the audit table to the admin user grant all on to strmadmin; Connect as the streams admin user and create a monitor table conn strmadmin/strmadmin CREATE TABLE streams_monitor ( date_and_time TIMESTAMP(6) DEFAULT systimestamp, txt_msg CLOB );Configuring Streams Create the CDC queue BEGIN ( queue_table => ' ', queue_name => ' ');END;/Configuring Streams Create capture rules BEGIN ( table_name => ' ', streams_type => 'capture', streams_name => 'capture_emp', queue_name => ' ', include_dml => true, include_ddl => false, inclusion_rule => true);END;/Configuring Streams Add extra attributes BEGIN ( capture_name => 'capture_emp', attribute_name => 'username', include => true);END;/Configuring Streams Mark redo entry as starting point (SCN) DECLARE iscn NUMBER; BEGIN iscn := (); ( source_object_name => ' ', source_database_name => 'ORCL', instantiation_scn => iscn);END.

6 /Configuring Streams Create DML Handler Proc CREATE OR REPLACE PROCEDURE emp_dml_handler(in_any IN ANYDATA) IS lcr $_ROW_RECORD; rc PLS_INTEGER; command VARCHAR2(30); old_values $_ROW_LIST;BEGIN -- Access the LCR rc := (lcr); -- Get the object command type command := (); -- I am inserting the XML equivalent of the LCR into the monitoring table. insert into streams_monitor (txt_msg) values (command || (in_any) );Configuring Streams Create DML Handler (cont'd) -- Set the command_type in the row LCR to INSERT ('INSERT'); -- Set the object_name in the row LCR to EMP_DEL ('EMPLOYEE_AUDIT'); -- Set the new values to the old values for update and delete IF command IN ('DELETE', 'UPDATE') THEN -- Get the old values in the row LCR old_values := ('old'); -- Set the old values in the row LCR to the new values in the row LCR ('new', old_values); -- Set the old values in the row LCR to NULL ('old', NULL); END IF;Configuring Streams Create DML Handler (cont'd) -- Add a SYSDATE for upd_date ('new', 'UPD_DATE', (SYSDATE)).

7 -- Add a user column ('new', 'user_name', ('USERNAME') ); -- Add an action column ('new', 'ACTION', (command)); -- Make the changes (true); commit;END;/Configuring Streams Register DML Handler BEGIN ( object_name => ' ', object_type => 'TABLE', operation_name => 'INSERT', error_handler => false, user_procedure => ' ', apply_database_link => NULL, apply_name => NULL);END;/ Do the exact same for update and Streams Create the apply rule DECLARE emp_rule_name_dml VARCHAR2(30); emp_rule_name_ddl VARCHAR2(30);BEGIN ( table_name => ' ', streams_type => 'apply', streams_name => 'apply_emp', queue_name => ' ', include_dml => true, include_ddl => false, source_database => 'ORCL', dml_rule_name => emp_rule_name_dml, ddl_rule_name => emp_rule_name_ddl); ( rule_name => emp_rule_name_dml, destination_queue_name => ' ');END;/Configuring Streams Turn off disable on error, start the apply and capture processes BEGIN ( apply_name => 'apply_emp', parameter => 'disable_on_error', value => 'n');END;/ BEGIN ( apply_name => 'apply_emp');END;/ BEGIN ( capture_name => 'capture_emp');END.

8 /Configuring Streams Test Insert a record into Select from to see audit records Select from streams_monitor to see LCR informationConfiguring Streams Configure 9i instance (SECOND) sqlplus / as sysdba create tablespace streams_second_tbs datafile 'c:\temp\ ' size 25M reuse autoextend on maxsize unlimited; create user strmadmin identified by strmadmin default tablespace streams_second_tbs quota unlimited on streams_second_tbs; grant dba to strmadmin;Configuring Streams Create the Audit Table in the 9i instance CREATE TABLE employee_audit( employee_id NUMBER(6), first_name VARCHAR2(20), last_name VARCHAR2(25), email VARCHAR2(25), phone_number VARCHAR2(20), hire_date DATE, job_id VARCHAR2(10), salary NUMBER(8,2), commission_pct NUMBER(2,2), manager_id NUMBER(6), department_id NUMBER(4), upd_date DATE, user_name VARCHAR2(30), action VARCHAR2(30));Configuring Streams Create the AQ queue in SECOND BEGIN ( queue_table => 'lrc_emp_t', queue_payload_type => ' ', multiple_consumers => TRUE, compatible => ' '); ( queue_name => 'lrc_emp_q', queue_table => 'lrc_emp_t').

9 ( queue_name => 'lrc_emp_q');END;/Configuring Streams Create a link from SECOND to ORCL CREATE Database LINK CONNECT TO strmadmin IDENTIFIED BY strmadmin USING ' '; Create a link from ORCL to SECOND CREATE Database LINK CONNECT TO strmadmin IDENTIFIED BY strmadmin USING ' ';Configuring Streams In 10g (ORCL), as strmadmin, create a propagation schedule for the captured data BEGIN ( table_name => ' ', streams_name => 'orcl_2_second', source_queue_name => ' ', destination_queue_name => include_dml => true, include_ddl => FALSE, source_database => ' ');END;/Configuring Streams Modify the DML Handler Proc (add this just before the procedure). DECLARE enqueue_options ; message_properties ; message_handle RAW(16); recipients $_recipient_list_t; BEGIN recipients(1) := $_agent( 'anydata_subscriber', NULL); := recipients;Configuring Streams Modify the DML Handler Proc (in 10g) ( queue_name => ' ', enqueue_options => enqueue_options, message_properties => message_properties, payload => (lcr), msgid => message_handle); EXCEPTION WHEN OTHERS THEN insert into streams_monitor (txt_msg) values ('Anydata: ' || ); END.

10 Configuring Streams Create a DEQUEUE proc in 9i CREATE OR REPLACE PROCEDURE emp_dq (consumer IN VARCHAR2) AS msg ANYDATA; row_lcr $_ROW_RECORD; num_var pls_integer; more_messages BOOLEAN := true; navigation VARCHAR2(30); BEGIN navigation := 'FIRST MESSAGE'; WHILE (more_messages) LOOP ('Looping thru messages'); BEGIN ( queue_name => ' ', streams_name => consumer, payload => msg, navigation => navigation, wait => );Configuring Streams Create a DEQUEUE proc in 9i IF () = ' $_ROW_RECORD' THEN num_var := (row_lcr); ( || ' row LCR dequeued'); END IF; navigation.