关于Change Data Capture（二）
作者：NinGoo | 【转载须以超链接形式标明文章原始出处和作者信息】
在关于Change Data Capture（一）中介绍了CDC的一些基本概念和类型。这篇文章主要是通过一个实际的例子来演示实现同步模式的CDC的基本步骤。
一.版本
SYS@ning>select * from v$version;
BANNER
-------------------------------------------------------------------------
Oracle Database 10g Enterprise Edition Release 10.2.0.3.0 - Prod
PL/SQL Release 10.2.0.3.0 - Production
CORE 10.2.0.3.0 Production
TNS for 32-bit Windows: Version 10.2.0.3.0 - Production
NLSRTL Version 10.2.0.3.0 - Production
BANNER
-------------------------------------------------------------------------
Oracle Database 10g Enterprise Edition Release 10.2.0.3.0 - Prod
PL/SQL Release 10.2.0.3.0 - Production
CORE 10.2.0.3.0 Production
TNS for 32-bit Windows: Version 10.2.0.3.0 - Production
NLSRTL Version 10.2.0.3.0 - Production
二.设置发布者
1.首先在source database创建一个用户作为发布者
SYS@ning>create user cdcpub identified by cdcpub;
User created.
User created.
2.授予相应的权限
SYS@ning>grant execute_catalog_role to cdcpub;
Grant succeeded.
SYS@ning>grant select_catalog_role to cdcpub;
Grant succeeded.
SYS@ning>grant create table to cdcpub;
Grant succeeded.
SYS@ning>grant create session to cdcpub;
Grant succeeded.
SYS@ning>grant execute on dbms_cdc_publish to cdcpub;
Grant succeeded.
Grant succeeded.
SYS@ning>grant select_catalog_role to cdcpub;
Grant succeeded.
SYS@ning>grant create table to cdcpub;
Grant succeeded.
SYS@ning>grant create session to cdcpub;
Grant succeeded.
SYS@ning>grant execute on dbms_cdc_publish to cdcpub;
Grant succeeded.
三.设置初始化参数
同步CDC，需要将java_pool_size设置为合适的大小，估计是其内部是采用java存储过程来实现的。
SYS@ning>alter system set java_pool_size=48M;
System altered.
System altered.
四.发布变化数据
1.例如要发布用户ning下的sales表
SYS@ning>desc ning.sales;
Name Null? Type
------------------------------ -------- ----------------
ID NUMBER(38)
PRODUCTID NUMBER(38)
PRICE NUMBER(10,2)
QUANTITY NUMBER(38)
Name Null? Type
------------------------------ -------- ----------------
ID NUMBER(38)
PRODUCTID NUMBER(38)
PRICE NUMBER(10,2)
QUANTITY NUMBER(38)
2.授予cdcpub用户对于该表的权限
SYS@ning>grant all on ning.sales to cdcpub;
Grant succeeded.
Grant succeeded.
3.创建chang set
SYS@ning>begin
2 dbms_cdc_publish.create_change_set(
3 change_set_name =>'ning_sales',
4 description =>'change set for ning.sales',
5 change_source_name =>'SYNC_SOURCE');
6 end;
7 /
PL/SQL procedure successfully completed.
2 dbms_cdc_publish.create_change_set(
3 change_set_name =>'ning_sales',
4 description =>'change set for ning.sales',
5 change_source_name =>'SYNC_SOURCE');
6 end;
7 /
PL/SQL procedure successfully completed.
同步CDC的chang source必须是SYNC_SOURCE。
4.创建change table
如果要将change table创建到users表空间
SYS@ning>alter user cdcpub quota unlimited on users;
User altered.
SYS@ning>begin
2 dbms_cdc_publish.create_change_table(
3 owner =>'cdcpub',
4 change_table_name=>'sales_ct',
5 change_set_name=>'ning_sales',
6 source_schema=>'ning',
7 source_table=>'sales',
8 column_type_list=>'id int,productid int,price number(10,2),quantity int',
9 capture_values=>'both',
10 rs_id=>'y',
11 row_id=>'n',
12 user_id=>'n',
13 timestamp=>'n',
14 object_id=>'n',
15 source_colmap=>'y',
16 target_colmap=>'y',
17 options_string=>'tablespace users');
18 end;
19 /
PL/SQL procedure successfully completed.
User altered.
SYS@ning>begin
2 dbms_cdc_publish.create_change_table(
3 owner =>'cdcpub',
4 change_table_name=>'sales_ct',
5 change_set_name=>'ning_sales',
6 source_schema=>'ning',
7 source_table=>'sales',
8 column_type_list=>'id int,productid int,price number(10,2),quantity int',
9 capture_values=>'both',
10 rs_id=>'y',
11 row_id=>'n',
12 user_id=>'n',
13 timestamp=>'n',
14 object_id=>'n',
15 source_colmap=>'y',
16 target_colmap=>'y',
17 options_string=>'tablespace users');
18 end;
19 /
PL/SQL procedure successfully completed.
创建好的change table定义如下
SYS@ning>desc cdcpub.sales_ct;
Name Null? Type
----------------------------------------- -------- ----------------------------
OPERATION$ CHAR(2)
CSCN$ NUMBER
COMMIT_TIMESTAMP$ DATE
RSID$ NUMBER
SOURCE_COLMAP$ RAW(128)
TARGET_COLMAP$ RAW(128)
ID NUMBER(38)
PRODUCTID NUMBER(38)
PRICE NUMBER(10,2)
QUANTITY NUMBER(38)
Name Null? Type
----------------------------------------- -------- ----------------------------
OPERATION$ CHAR(2)
CSCN$ NUMBER
COMMIT_TIMESTAMP$ DATE
RSID$ NUMBER
SOURCE_COLMAP$ RAW(128)
TARGET_COLMAP$ RAW(128)
ID NUMBER(38)
PRODUCTID NUMBER(38)
PRICE NUMBER(10,2)
QUANTITY NUMBER(38)
5.将change table的读取权限授予订阅者
先创建一个订阅者帐号
SYS@ning>create user cdcsub identified by cdcsub;
User created.
SYS@ning>grant create session to cdcsub;
Grant succeeded.
SYS@ning>grant create table to cdcsub;
Grant succeeded.
SYS@ning>grant select on cdcpub.sales_ct to cdcsub;
Grant succeeded.
User created.
SYS@ning>grant create session to cdcsub;
Grant succeeded.
SYS@ning>grant create table to cdcsub;
Grant succeeded.
SYS@ning>grant select on cdcpub.sales_ct to cdcsub;
Grant succeeded.
如果这里没有将change table授权给订阅者，那么在后面订阅的时候，就会报错:
ORA-31466: no publications found
ORA-06512: at “SYS.DBMS_CDC_SUBSCRIBE”, line 19
ORA-06512: at line 2
五.订阅变化数据
1.查看可以订阅的源数据表
SYS@ning>conn cdcsub/cdcsub;
Connected.
CDCSUB@ning>select * from all_source_tables;
SOURCE_SCHEMA_NAME SOURCE_TABLE_NAME
------------------------------ ------------------------------
NING SALES
Connected.
CDCSUB@ning>select * from all_source_tables;
SOURCE_SCHEMA_NAME SOURCE_TABLE_NAME
------------------------------ ------------------------------
NING SALES
2.查看可以订阅的change set和column
CDCSUB@ning>select distinct change_set_name,column_name,pub_id
2 from all_published_columns
3 where source_schema_name='NING' and source_table_name='SALES';
CHANGE_SET COLUMN_NAME PUB_ID
---------- -------------------- ----------
NING_SALES ID 52590
NING_SALES PRICE 52590
NING_SALES PRODUCTID 52590
NING_SALES QUANTITY 52590
2 from all_published_columns
3 where source_schema_name='NING' and source_table_name='SALES';
CHANGE_SET COLUMN_NAME PUB_ID
---------- -------------------- ----------
NING_SALES ID 52590
NING_SALES PRICE 52590
NING_SALES PRODUCTID 52590
NING_SALES QUANTITY 52590
3.创建订阅
CDCSUB@ning>begin
2 dbms_cdc_subscribe.create_subscription(
3 change_set_name=>'ning_sales',
4 description=>'change data for sales',
5 subscription_name=>'sales_sub');
6 end;
7 /
PL/SQL procedure successfully completed.
2 dbms_cdc_subscribe.create_subscription(
3 change_set_name=>'ning_sales',
4 description=>'change data for sales',
5 subscription_name=>'sales_sub');
6 end;
7 /
PL/SQL procedure successfully completed.
4.订阅具体的source table和column
一个订阅可以同时包含同一个change set中的多个source table
CDCSUB@ning>begin
2 dbms_cdc_subscribe.subscribe(
3 subscription_name=>'sales_sub',
4 source_schema=>'NING',
5 source_table=>'SALES',
6 column_list=>'id,productid,price,quantity',
7 subscriber_view=>'sales_view');
8 end;
9 /
PL/SQL procedure successfully completed.
2 dbms_cdc_subscribe.subscribe(
3 subscription_name=>'sales_sub',
4 source_schema=>'NING',
5 source_table=>'SALES',
6 column_list=>'id,productid,price,quantity',
7 subscriber_view=>'sales_view');
8 end;
9 /
PL/SQL procedure successfully completed.
订阅会创建一个叫做sales_view的订阅视图，通过该视图，我们可以获得源表的变化数据。
CDCSUB@ning>desc sales_view;
Name Null? Type
----------------------------------------- -------- ----------------------------
OPERATION$ CHAR(2)
CSCN$ NUMBER
COMMIT_TIMESTAMP$ DATE
RSID$ NUMBER
SOURCE_COLMAP$ RAW(128)
TARGET_COLMAP$ RAW(128)
ID NUMBER(38)
PRICE NUMBER(10,2)
PRODUCTID NUMBER(38)
QUANTITY NUMBER(38)
Name Null? Type
----------------------------------------- -------- ----------------------------
OPERATION$ CHAR(2)
CSCN$ NUMBER
COMMIT_TIMESTAMP$ DATE
RSID$ NUMBER
SOURCE_COLMAP$ RAW(128)
TARGET_COLMAP$ RAW(128)
ID NUMBER(38)
PRICE NUMBER(10,2)
PRODUCTID NUMBER(38)
QUANTITY NUMBER(38)
5.激活订阅
不管订阅包含一个source table还是多个，只需要执行一次激活即可。
CDCSUB@ning>begin
2 dbms_cdc_subscribe.activate_subscription(
3 subscription_name=>'SALES_SUB');
4 end;
5 /
PL/SQL procedure successfully completed.
2 dbms_cdc_subscribe.activate_subscription(
3 subscription_name=>'SALES_SUB');
4 end;
5 /
PL/SQL procedure successfully completed.
6.扩展订阅窗口
在源表数据变化后，变化的数据在订阅端需要执行extend_window后才能看见
CDCSUB@ning>begin
2 dbms_cdc_subscribe.extend_window(
3 subscription_name=>'SALES_SUB');
4 end;
5 /
PL/SQL procedure successfully completed.
2 dbms_cdc_subscribe.extend_window(
3 subscription_name=>'SALES_SUB');
4 end;
5 /
PL/SQL procedure successfully completed.
如果是第一次执行该过程，则会在订阅段看到所有的变化数据，以后每次执行都只增加到从上次执行以来的新数据，相当于物化视图的增量刷新。
7.查询订阅视图中的数据
CDCSUB@ning>select id,productid,price,quantity from sales_view;
ID PRODUCTID PRICE QUANTITY
---------- ---------- ---------- ----------
1 1 20.2 100
2 1 20.2 200
3 2 40.8 50
ID PRODUCTID PRICE QUANTITY
---------- ---------- ---------- ----------
1 1 20.2 100
2 1 20.2 200
3 2 40.8 50
9.清除当前窗口中的变化数据
如果当前变化数据已经不再需要，可以清除其数据
CDCSUB@ning>begin
2 dbms_cdc_subscribe.purge_window(
3 subscription_name=>'SALES_SUB');
4 end;
5 /
PL/SQL procedure successfully completed.
2 dbms_cdc_subscribe.purge_window(
3 subscription_name=>'SALES_SUB');
4 end;
5 /
PL/SQL procedure successfully completed.
清除后，在订阅视图中就看不到之前的变化数据了
CDCSUB@ning>select id,productid,price,quantity from sales_view;
no rows selected
no rows selected
10.停止订阅
CDCSUB@ning>begin
2 dbms_cdc_subscribe.drop_subscription(
3 subscription_name=>'SALES_SUB');
4 end;
5 /
PL/SQL procedure successfully completed.
2 dbms_cdc_subscribe.drop_subscription(
3 subscription_name=>'SALES_SUB');
4 end;
5 /
PL/SQL procedure successfully completed.
订阅视图也同时被删除
CDCSUB@ning>select id,productid,price,quantity from sales_view;
select id,productid,price,quantity from sales_view
*
ERROR at line 1:
ORA-00942: table or view does not exist
select id,productid,price,quantity from sales_view
*
ERROR at line 1:
ORA-00942: table or view does not exist