oracle stream Step by Step

yxtg

环境设置：
set echo on;
ACCEPT dba_pwd_src PROMPT 'Enter Password of user "sys" to create Streams Admin at Source : '   
ACCEPT strm_pwd_src PROMPT 'Enter Password of Streams Admin "STRMADMIN" to be created at Source : '   
ACCEPT dba_pwd_dest PROMPT 'Enter Password of user "sys" to create Streams Admin at Destination : '   
ACCEPT strm_pwd_dest PROMPT 'Enter Password of Streams Admin "STRMADMIN" to be created at Destination : '  
connect "SYS"/&dba_pwd_src as SYSDBA;
alter user "STRMADMIN" identified by &strm_pwd_src;
grant DBA, IMP_FULL_DATABASE, EXP_FULL_DATABASE to  "STRMADMIN";
BEGIN
  DBMS_STREAMS_AUTH.GRANT_ADMIN_PRIVILEGE(
    grantee => '"STRMADMIN"',
    grant_privileges => true);
END;
/
COMMIT;

connect  "STRMADMIN"/&strm_pwd_src;
CREATE DATABASE LINK RECOVER.GYIC.COM connect to  "STRMADMIN" identified by "STRMADMIN" using '  (DESCRIPTION =
    (ADDRESS = (PROTOCOL = TCP)(HOST = 9.8.0.201)(PORT = 1521))
    (CONNECT_DATA =
      (SERVER = DEDICATED)
      (SERVICE_NAME = gydb.recover)
    )
  )';
  
  
BEGIN
  DBMS_STREAMS_ADM.SET_UP_QUEUE(
    queue_table => '"STREAMS_CAPTURE1_QT"',
    queue_name  => '"STREAMS_CAPTURE1_Q"',
    queue_user  => '"STRMADMIN"');
END;
/
COMMIT;

BEGIN
  DBMS_STREAMS_ADM.ADD_SCHEMA_RULES(
    schema_name        => '"GYIC"',
    streams_type       => 'capture',
    streams_name       => '"STREAMS_CAPTURE1"',
    queue_name         => '"STRMADMIN"."STREAMS_CAPTURE1_Q"',
    include_dml        => true,
    include_ddl        => false,
    include_tagged_lcr => false,
    inclusion_rule     => true);
END;
/

BEGIN
  DBMS_STREAMS_ADM.ADD_SCHEMA_PROPAGATION_RULES(
   schema_name            => '"GYIC"',
   streams_name           => '"STREAMS_PROPAGATION1"',
    source_queue_name      => '"STRMADMIN"."STREAMS_CAPTURE1_Q"',
    destination_queue_name => '"STRMADMIN"."STREAMS_APPLY1_Q"@RECOVER.GYIC.COM',
    include_dml            => true,
    include_ddl            => false,
    source_database        => 'GYDB.GYIC.COM',
    inclusion_rule         => true );
END;
/
COMMIT;

connect   "SYS"/&dba_pwd_dest@"(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)(HOST=9.8.0.201)(PORT=1521)))(CONNECT_DATA=(SERVICE_NAME=GYDB.RECOVER)(SERVER=DEDICATED)))" as SYSDBA;

alter user  "STRMADMIN" identified by &strm_pwd_dest;
grant DBA, IMP_FULL_DATABASE, EXP_FULL_DATABASE to "STRMADMIN";

BEGIN
  DBMS_STREAMS_AUTH.GRANT_ADMIN_PRIVILEGE(
    grantee => '"STRMADMIN"',
    grant_privileges => true);
END;
/
COMMIT;

connect "STRMADMIN"/&strm_pwd_dest@"(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)(HOST=9.8.0.201)(PORT=1521)))(CONNECT_DATA=(SID=RECOVER)(SERVER=DEDICATED)))";

CREATE DATABASE LINK GYDB.GYIC.COM connect to "STRMADMIN" identified by &strm_pwd_src using '(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)(HOST=ed-olraclin50d)(PORT=1521)))(CONNECT_DATA=(SID=GYDB)(server=DEDICATED)))';
BEGIN
  DBMS_STREAMS_ADM.SET_UP_QUEUE(
    queue_table => '"STREAMS_APPLY1_QT"',
    queue_name  => '"STREAMS_APPLY1_Q"',
    queue_user  => '"STRMADMIN"');
END;
/
BEGIN
  DBMS_STREAMS_ADM.ADD_SCHEMA_RULES(
    schema_name        => '"GYIC"',
    streams_type       => 'apply',
    streams_name       => '"STREAMS_APPLY1"',
    queue_name         => '"STRMADMIN"."STREAMS_APPLY1_Q"',
    include_dml        => true,
    include_ddl        => false,
    include_tagged_lcr => false,
    inclusion_rule     => true);
END;
/

导入数据：

ACCEPT strm_pwd_src PROMPT 'Enter Password of Streams Admin "STRMADMIN" at Source : ' HIDE  
ACCEPT strm_pwd_dest PROMPT 'Enter Password of Streams Admin "STRMADMIN" at Destination : ' HIDE  
connect "STRMADMIN"/&strm_pwd_dest@"(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)(HOST=9.8.0.201)(PORT=1521)))(CONNECT_DATA=(SID=RECOVER)(SERVER=DEDICATED)))";
set serverout on;
DECLARE
  handle1 number;
  ind number;
  percent_done number;
  job_state VARCHAR2(30);
  le ku$_LogEntry;
  js ku$_JobStatus;
  jd ku$_JobDesc;
  sts ku$_Status;
BEGIN   
  handle1 := DBMS_DATAPUMP.OPEN('IMPORT','SCHEMA', 'GYDB.GYIC.COM');
  DBMS_DATAPUMP.ADD_9.8.0.201(handle1, 'StreamImport_1240813852603.log', 'DATA_PUMP_DIR', '',  DBMS_DATAPUMP.KU$_9.8.0.201_TYPE_LOG_9.8.0.201);
  DBMS_DATAPUMP.SET_PARAMETER(handle1, 'FLASHBACK_SCN', 117319133);
   DBMS_DATAPUMP.METADATA_FILTER(handle1, 'SCHEMA_EXPR', 'IN (''GYIC'')');
  DBMS_DATAPUMP.SET_PARAMETER(handle1, 'INCLUDE_METADATA', 1);
  DBMS_DATAPUMP.START_JOB(handle1);  
  percent_done :=0;
  job_state := 'UNDEFINED';
  while (job_state != 'COMPLETED') and (job_state != 'STOPPED') loop
  dbms_datapump.get_status(handle1, dbms_datapump.ku$_status_job_error + dbms_datapump.ku$_status_job_status + dbms_datapump.ku$_status_wip,-1,job_state,sts);
  js := sts.job_status;
  if js.percent_done != percent_done
  then
     dbms_output.put_line('*** Job percent done = ' || to_char(js.percent_done));
     percent_done := js.percent_done;
  end if;
  if(bitand(sts.mask, dbms_datapump.ku$_status_wip) != 0)
  then
    le := sts.wip;
  else
     if(bitand(sts.mask,dbms_datapump.ku$_status_job_error) != 0)
     then
       le := sts.error;
     else
       le := null;
     end if;
  end if;
  if le is not null
  then
    ind := le.FIRST;
    while ind is not null loop
      dbms_output.put_line(le(ind).LogText);
      ind := le.NEXT(ind);
    end loop;
  end if;
  end loop;
  dbms_output.put_line('Job has completed');
  dbms_output.put_line('Final job state = ' || job_state);
  dbms_datapump.detach(handle1);
END;   
/、

启动服务：
ACCEPT strm_pwd_src PROMPT 'Enter Password of Streams Admin "STRMADMIN" at Source : ' HIDE  
ACCEPT strm_pwd_dest PROMPT 'Enter Password of Streams Admin "STRMADMIN" at Destination : ' HIDE  
connect "STRMADMIN"/&strm_pwd_dest@"(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)(HOST=9.8.0.201)(PORT=1521)))(CONNECT_DATA=(SID=RECOVER)(SERVER=DEDICATED)))";
set serverout on;
BEGIN
DBMS_APPLY_ADM.SET_SCHEMA_INSTANTIATION_SCN(
    source_schema_name   => '"GYIC"',
   source_database_name => 'GYDB.GYIC.COM',
   instantiation_scn    => 117319133,
   recursive            => true);
END;
/
DECLARE
   v_started number;
BEGIN
SELECT DECODE(status, 'ENABLED', 1, 0) INTO v_started
FROM DBA_APPLY where apply_name = 'STREAMS_APPLY1';
if (v_started = 0) then
  DBMS_APPLY_ADM.START_APPLY(apply_name => '"STREAMS_APPLY1"');
end if;
END;
/
connect "STRMADMIN"/&strm_pwd_src;

set serverout on;
DECLARE
   v_started number;
BEGIN
SELECT DECODE(status, 'ENABLED', 1, 0) INTO v_started
FROM DBA_CAPTURE where CAPTURE_NAME = 'STREAMS_CAPTURE1';
if (v_started = 0) then
  DBMS_CAPTURE_ADM.START_CAPTURE(capture_name => '"STREAMS_CAPTURE1"');
end if;
END;
/
BEGIN
DBMS_OUTPUT.PUT_LINE('*** Progress Message ===> Started the capture process STREAMS_CAPTURE1 at source database GYDB and the apply process STREAMS_APPLY1 at the destination database successfully. ***');
END;
/