关键字：

电科金仓、KingbaseFlySync、KFS、同步程序、replicator、数据同步、数据解析、增量同步、实时同步、DML过滤、操作类型过滤DML、insert语句过滤、update语句过滤、delete语句过滤

1.功能介绍

实时同步支持基于操作类型（insert/update/delete）对DML语句进行过滤。

2.使用场景

KingbaseFlySync在实时同步的过程中默认同步属于同步范围内表所有insert/update/delete操作产生的增量数据，不对数据进行过滤。在特定场景下，用户希望只同步某些操作的DML，基于操作类型过滤可以实现指定操作类型对INSERT/DELETE/UPDATE操作的DML数据进行过滤。

3.配置使用方式介绍

3.1.配置方式

需要在同步服务配置文件flysync.ini中添加对应的过滤参数并指定操作类型过滤规则文件，具体配置参数如下所示

svc-extractor-filters=skipeventbytype
property=replicator.filter.skipeventbytype.definitionsFile=${replicator.home.dir}/support/filters-config/skipeventbytype.json

3.2.操作类型过滤规则文件配置方法

操作类型过滤规则文件skipeventbytype.json的配置示例如下：
1、 场景一： 跳过所有模式所有表的 INSERT、DELETE、UPDATE 操作

{
    "*": {
        "*": "INSERT,UPDATE,DELETE"
    }
}

2、 场景二： 跳过指定模式指定表的 INSERT/DELETE/UPDATE 操作

{
    "SCHEMA_NAME_1": {
        "TABLE_NAME_11": "DELETE",          /* 过滤掉 DELETE 操作 */
        "TABLE_NAME_12": "UPDATE,DELETE"  /* 过滤掉 UPDATE和DELETE 操作 */
        "TABLE_NAME_13": "UPDATE,DELETE"  /* 过滤掉 UPDATE和DELETE 操作 */
    },
    "SCHEMA_NAME_2": {
        "TABLE_NAME_21": "DELETE",         /* 过滤掉 DELETE 操作 */
        "TABLE_NAME_22": "UPDATE,DELETE"  /* 过滤掉 UPDATE和DELETE 操作 */
    }
}

3、 场景三： 跳过 匹配通配符模式 下 匹配通配符表 的 INSERT/DELETE/UPDATE 操作

{
/* 过滤掉 所有 模式下的 TABLE_NAME 表的 DELETE 操作 */
    "*": {
        "TABLE_NAME": "DELETE"
},
/* 过滤掉 所有 模式下的以_TABLE_NAME结尾的表的 DELETE 操作 */
    "*": {
        "*_TABLE_NAME": "DELETE"
},
/* 过滤掉 所有 模式下的以TABLE_NAME_开头的表的 DELETE 操作 */
    "*": {
        "TABLE_NAME_*": "DELETE"
    }
}

4、 场景四： 跳过指定模式下匹配通配符表的 INSERT/DELETE/UPDATE 操作

{
    /* 过滤掉 SCHEMA_NAME_1 模式下以 TABLE_NAME_ 开头的表的 DELETE 操作 */
    "SCHEMA_NAME_1": {
        "TABLE_NAME_*": "DELETE"
},
/* 过滤掉 SCHEMA_NAME_2 模式下以 TABLE_NAME_ 结尾的表的 DELETE 操作 */
    "SCHEMA_NAME_2": {
        "*_TABLE_NAME": "DELETE"
},
/* 过滤掉 SCHEMA_NAME_3 模式下的所有表的 DELETE 操作 */
    "SCHEMA_NAME_3": {
        "*": "DELETE"
    }
}

4.示例

4.1.前置条件

1）完成安装KFS同步程序前的环境准备；
2）源端目标端数据库含有以下表结构：

create table MYTEST.T1 (ID int,C1 varchar2(100));
create table MYTEST.T2 (ID int,C1 varchar2(100));
create table TPCC.T1 (ID int,C1 varchar2(100));
create table TPCC.T2 (ID int,C1 varchar2(100));

4.2.操作步骤

1）配置flysync.ini文件，在源端的flysync.ini中添加以下参数

[default]
...

[source]
...
svc-extractor-filters=skipeventbytype   #添加基于操作类型的过滤器
property=replicator.filter.skipeventbytype.definitionsFile=/home/oracle/skipeventbytype.json   

#指定基于操作类型的过滤规则文件

2）配置基于操作类型的过滤规则的skipeventbytype.json文件，内容如下：

{
    "MYTEST" : {
        "T1" : "UPDATE",   /*过滤掉 MYTEST模式下的T1表的UPDATE 操作 */
        "T2" : "UPDATE,DELETE"  /* 过滤掉MYTEST模式下的T2表的UPDATE和DELETE 操作 */
    },
    "TPCC" : {
        "T1" : "DELETE"   /* 过滤掉 TPCC模式下的T1表的DELETE 操作*/
    }
}

3）安装同步程序并启动。具体的安装步骤参见《Kingbase FlySync 安装部署手册》
4）在源端数据库执行以下操作：

insert into MYTEST.T1 values (1,'mytest_t1_ceshi001');
insert into MYTEST.T1 values (2,'mytest_t1_ceshi002');
update MYTEST.T1 set C1='mytest_t1_update002' where id=1;
delete from MYTEST.T1 where id=2;

查看kufl 可以看到MYTEST.T1的update操作产生的增量数据被过滤了

查看目标端的数据，update操作没有同步至目标端，被过滤了：

5）在源端执行以下操作：

insert into MYTEST.T2 values (1,'mytest_t2_ceshi001');
insert into MYTEST.T2 values (2,'mytest_t2_ceshi002');
update MYTEST.T2 set C1='mytest_t2_update002' where id=1;
delete from MYTEST.T2 where id=2;

查看kufl，可以看到MYTEST.T2表的update、delete操作都被过滤了

查看目标端的数据，update、delete操作都没有同步至目标端，被过滤了：

6）在源端执行以下操作：

insert into TPCC.T1 values (1,'tpcc_t1_ceshi001');
insert into TPCC.T1 values (2,'tpcc_t1_ceshi002');
update TPCC.T1 set C1='tpcc_t1_update002' where id=1;
delete from TPCC.T1 where id=2;

查看kufl，可以看到TPCC.T1表的delete操作都被过滤了

查看目标端的数据，delete操作都没有同步至目标端，被过滤了：

7）在源端执行以下操作：

insert into TPCC.T2 values (1,'tpcc_t2_ceshi001');
insert into TPCC.T2 values (2,'tpcc_t2_ceshi002');
update TPCC.T2 set C1='tpcc_t2_update002' where id=1;
delete from TPCC.T2 where id=2;

查看kufl，可以看到TPCC.T2表所有的操作都被解析出来了：

查看目标端的数据，所有的操作都同步到了目标端：