配置ORACLE STREAMS TRANSFORMATION实现异构表的同步

数据库版本：ORACLE 10.2.0.1

测试环境：

源库：stream76.com

目的库：stream19.com

源表：

test.t11

 Name                                      Null?    Type

 ----------------------------------------- -------- ----------------------------

 ID                                                 NUMBER(38)

目标表：

test.t11_19

 Name                                      Null?    Type

 ----------------------------------------- -------- ----------------------------

 NO                                                 NUMBER(38)

1 在所有库创建用户并授权

connect / as sysdba

CREATE USER strmadmin IDENTIFIED BY strmadmin DEFAULT TABLESPACE data_01 QUOTA UNLIMITED ON tbs_repadmin;

GRANT DBA, CONNECT, RESOURCE, SELECT_CATALOG_ROLE TO strmadmin;

GRANT EXECUTE ON DBMS_AQADM TO strmadmin;

GRANT EXECUTE ON DBMS_CAPTURE_ADM TO strmadmin;

GRANT EXECUTE ON DBMS_PROPAGATION_ADM TO strmadmin;

GRANT EXECUTE ON DBMS_STREAMS_ADM TO strmadmin;

GRANT EXECUTE ON DBMS_APPLY_ADM TO strmadmin;

GRANT EXECUTE ON DBMS_FLASHBACK TO strmadmin;

BEGIN 

    DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(

        privilege => DBMS_RULE_ADM.CREATE_RULE_SET_OBJ, 

        grantee => 'strmadmin', 

        grant_option => FALSE);

END;

/

BEGIN 

    DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(

        privilege => DBMS_RULE_ADM.CREATE_RULE_OBJ, 

        grantee => 'strmadmin', 

        grant_option => FALSE);

END;

/

execute DBMS_STREAMS_AUTH.GRANT_ADMIN_PRIVILEGE('STRMADMIN');

2 源库：创建DBLINK和队列：

connect strmadmin/strmadmin

create database link STREAM19.COM

connect to strmadmin

identified by strmadmin

using '(DESCRIPTION=(ADDRESS = (PROTOCOL = TCP)(HOST = *.*.*.*)(PORT = 1521))(CONNECT_DATA=(SERVER = DEDICATED)(SID=bill)))';

BEGIN                                       

     DBMS_STREAMS_ADM.SET_UP_QUEUE(         

      queue_table => 'STREAMS_QUEUE_TABLE', 

      queue_name => 'STREAMS_QUEUE',        

      queue_user => 'STRMADMIN');           

END;                                        

/                                           

3 目标库：创建DBLINK和队列：

connect strmadmin/strmadmin

create database link STREAM76.COM

connect to strmadmin

identified by strmadmin

using '(DESCRIPTION=(ADDRESS = (PROTOCOL = TCP)(HOST = *.*.*.*)(PORT = 1521))(CONNECT_DATA=(SERVER = DEDICATED)(SID=bill)))';

BEGIN                                       

     DBMS_STREAMS_ADM.SET_UP_QUEUE(         

      queue_table => 'STREAMS_QUEUE_TABLE', 

      queue_name => 'STREAMS_QUEUE',        

      queue_user => 'STRMADMIN');           

END;                                        

/                                           

4 创建表应用规则

BEGIN

  DBMS_STREAMS_ADM.ADD_TABLE_RULES(table_name         => 'TEST.T11',

                                   streams_type       => 'apply',

                                   streams_name       => 'STREAMS_APPLY',

                                   queue_name         => 'STRMADMIN.STREAMS_QUEUE',

                                   include_dml        => true,

                                   include_ddl        => true,

                                   include_tagged_lcr => false,

                                   source_database    => 'STREAM76.COM',

                                   inclusion_rule     => true);

END;

/

5 配置复制时应用的转换，本处为列名转换和表名转换，并启动应用进程；

BEGIN 

  DBMS_STREAMS_ADM.RENAME_COLUMN(

    rule_name         => 'T1147',

    table_name        => 'TEST.T11',

    from_column_name  => 'id', 

    to_column_name    => 'no',

    value_type        => '*',

    step_number       => 0,

    operation         => 'ADD');

END;

/

begin

DBMS_STREAMS_ADM.RENAME_TABLE(

   rule_name        => 'T1147',

   from_table_name  => 'TEST.T11',

   to_table_name    => 'TEST.T11_19',

   step_number      => 0,

   operation        => 'ADD');

end;

/

BEGIN                                   

     DBMS_APPLY_ADM.ALTER_APPLY(        

      apply_name => 'STREAMS_APPLY',  

      apply_user => 'strmadmin');        

END;                                    

/                                       

DECLARE                                                                

   v_started number;                                                   

BEGIN                                                                  

   SELECT decode(status, 'ENABLED', 1, 0) INTO v_started               

   FROM DBA_APPLY WHERE APPLY_NAME = 'STREAMS_APPLY';                

   if (v_started = 0) then                                             

     DBMS_APPLY_ADM.START_APPLY(apply_name  => 'STREAMS_APPLY');     

   end if;                                                             

END;                                                                   

/                                                                      

6 源库：设置LogMiner表空间

connect / as sysdba

CREATE TABLESPACE LOGMNRTS DATAFILE '/Data/oradata_10g/bill/logmnrts.dbf' SIZE 25M AUTOEXTEND ON

    MAXSIZE UNLIMITED;

BEGIN

     DBMS_LOGMNR_D.SET_TABLESPACE('LOGMNRTS');

END;

/

7 添加表捕获和传播规则

connect strmadmin/strmadmin

BEGIN

  DBMS_STREAMS_ADM.ADD_TABLE_RULES(table_name         => 'TEST.T11',

                                   streams_type       => 'capture',

                                   streams_name       => 'STREAMS_CAPTURE',

                                   queue_name         => 'STRMADMIN.STREAMS_QUEUE',

                                   include_dml        => true,

                                   include_ddl        => true,

                                   include_tagged_lcr => false,

                                   source_database    => NULL,

                                   inclusion_rule     => true);

END;

/

BEGIN

  DBMS_STREAMS_ADM.ADD_TABLE_PROPAGATION_RULES(table_name             => 'TEST.T11',

                                               streams_name           => 'STREAMS_PROPAGATION',

                                               source_queue_name      => 'STRMADMIN.STREAMS_QUEUE',

                                               destination_queue_name => 'STRMADMIN.STREAMS_QUEUE@STREAM19.COM',

                                               include_dml            => true,

                                               include_ddl            => true,

                                               include_tagged_lcr     => false,

                                               source_database        => 'STREAM76.COM',

                                               inclusion_rule         => true,

                                               queue_to_queue         => true);

END;

/

8 数据初始

exp USERID=SYSTEM/oracle WNER=test FILE=test.dmp OBJECT_CONSISTENT=Y STATISTICS = NONE   

传送到目标库

scp test.dmp 202.108.251.19:/home/oracle

目标库：

imp USERID=SYSTEM/oracle FULL=Y CONSTRAINTS=Y FILE=test.dmp buffer=102400000 IGNORE=Y COMMIT=Y STREAMS_INSTANTIATION=Y  

9 源库：启动CAPTURE进程

connect strmadmin/strmadmin

begin                                                                   

      DBMS_CAPTURE_ADM.START_CAPTURE(capture_name  => 'STREAMS_CAPTURE');

end;                                                                

/                                                                        

测试：在源库进行增删改操作，测试同步

源库：

SQL> insert into test.t11 values (99);

1 row created.

SQL> commit;

Commit complete.

SQL> select * from test.t11;

        ID

----------

        99

目的库：

SQL> select * from test.t11_19;

        NO

----------

        99

源库：

SQL> update test.t11 set id=888 where id=99;

1 row updated.

SQL> commit;

Commit complete.

SQL> select * from test.t11;

        ID

----------

       888

SQL>

目的库：

SQL> l

  1* select * from test.t11_19

SQL> /

        NO

----------

                    
                    

                                                    来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/9375/viewspace-490762/
                            ，如需转载，请注明出处，否则将追究法律责任。