SPL 实践：异构库的数据比对
异构库的数据比对是指分别在不同类型数据库中的两张逻辑结构相同的数据表，进行数据比较，找出差异。
问题描述
数据对比的难点在于异构数据库间的数据类型和数据库本身在处理数据上存在的差异，这会导致同样的数据，进出不同数据库会得到不同的结果。
常用的数据类型主要有数值、字符串、日期。下面我们以Oracle和Mysql为例，分别创建一个即包含这三种常见数据类型，同时逻辑结构又比较简单的同构表，来看看两者的差异，以及如何使用SPL进行异构库的数据比对。
Oracle建表：
CREATE TABLE "TEST001" (
"F1" NUMBER(2, 0),
"F2" VARCHAR2(100),
"F3" DATE,
"F4" BINARY_DOUBLE
)
Mysql建表：
CREATE TABLE `test001` (
`f1` int,
`f2` varchar(100),
`f3` datetime,
`f4` decimal(10,2)
)
数据入库
按上述结构，使用SPL创建测试数据并分别入库的代码如下：
A |
|
1 |
=rand@s(1) |
2 |
=10.new(~:f1,rands("qwertyuiopasdfghjklzxcvbnm",5):f2,datetime("2011-11-11 11:11:11"):f3,rand(99999)/100:f4) |
3 |
>A2(3).run(#2=null),A2(5).run(#2="") |
4 |
=connect@l("mysql8") |
5 |
=A4.update(A2,test001,f1,f2,f3,f4;f1) |
6 |
>A4.close() |
7 |
=connect@l("oracle12c") |
8 |
=A7.update(A2,test001,f1,f2,f3,f4;f1) |
9 |
=A7.close() |
异构库差异
入库后，观察数据库中的数据，如下：
Oracle：
f1 |
f2 |
f3 |
f4 |
1 |
kqipd |
2011-11-11 11:11:11 |
156.82 |
2 |
xvvak |
2011-11-11 11:11:11 |
17.53 |
3 |
null |
2011-11-11 11:11:11 |
708.24 |
4 |
twtgq |
2011-11-11 11:11:11 |
825.2 |
5 |
null |
2011-11-11 11:11:11 |
628.82 |
6 |
hdjeb |
2011-11-11 11:11:11 |
3.12 |
7 |
zurnb |
2011-11-11 11:11:11 |
463.28 |
8 |
khcdv |
2011-11-11 11:11:11 |
903.9 |
9 |
dhdhn |
2011-11-11 11:11:11 |
956.73 |
10 |
vkacd |
2011-11-11 11:11:11 |
647.69 |
Myqsl：
f1 |
f2 |
f3 |
f4 |
1 |
kqipd |
2011-11-11 11:11:11 |
156.82 |
2 |
xvvak |
2011-11-11 11:11:11 |
17.53 |
3 |
null |
2011-11-11 11:11:11 |
708.24 |
4 |
twtgq |
2011-11-11 11:11:11 |
825.2 |
5 |
2011-11-11 11:11:11 |
628.82 |
|
6 |
hdjeb |
2011-11-11 11:11:11 |
3.12 |
7 |
zurnb |
2011-11-11 11:11:11 |
463.28 |
8 |
khcdv |
2011-11-11 11:11:11 |
903.9 |
9 |
dhdhn |
2011-11-11 11:11:11 |
956.73 |
10 |
vkacd |
2011-11-11 11:11:11 |
647.69 |
发现Oracle中，其中f1为5的记录，f2为null，而不是空串。这是因为在Oracle中空串(‘’)和null没有区别。
方法1：SQL中处理
把Mysql中的空串转为null后再比对，否则两边永远不会一致。在已知字段名的情况下，可以在SQL中直接转换，例如：select f1,if(f2='',null,f2) as f2,f3,f4 from test001 order by 1。
方法2：SPL中处理
当只知道表名，但不知道具体数据结构时，可以先通过SQL返回已知表的结构（列名、数据类型），再在SPL中拼出转换空串为null的SQL，代码如下：
A |
|
1 |
=connect@l("mysql8") |
2 |
=A1.query("SELECT COLUMN_NAME, DATA_TYPE FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME ='test001'") |
3 |
=A2.(if(#2=="varchar","if("/#1/"='',null,"/#1/") as "/#1,#1)).concat@c() |
4 |
=A1.query@x("select "/A3/" from test001") |
小数据量
当需要比对的数据量较小，可以装入内存时，找出两者的差异
有主键
以上述表结构，假设f1为主键，比对代码如下：
A |
|
1 |
=connect@l("mysql8") |
2 |
=A1.query@x("select f1,if(f2='',null,f2) as f2,f3,f4 from test001 order by 1") |
3 |
=connect@l("oracle12c") |
4 |
=A3.query@x("select * from test001 order by 1") |
5 |
=join@f(A2:t1,f1;A4:t2,f1) |
A5将A2与A4序表，按f1（主键）full join
但是观察结果，发现任何记录都无法相交。两边的f1都是1至10，应该都相等才对，这是什么原因呢？
这是因为SPL的join()时，默认采用的hash值比较，而Oracle中只有number类型。虽然根据DDL的number上设的参数可以限定数值范围（确定Java中对应的数据类型），但是Oracle的JDBC本身没处理，统一返回decimal类型。结果就导致两边的f1虽然值的大小一样，但是数据类型不一致，计算出的hash值也不同。
方法1:更改数据类型
A |
|
1 |
=connect@l("mysql8") |
2 |
=A1.query@x("select f1,if(f2='',null,f2) as f2,f3,f4 from test001 order by 1") |
3 |
=connect@l("oracle12c") |
4 |
=A3.query@x("select * from test001 order by 1").run(#1=int(#1)) |
5 |
=join@f(A2:t1,f1;A4:t2,f1) |
A4中将decimal类型f1的转为int类型
方法2:归并
A |
|
1 |
=connect@l("mysql8") |
2 |
=A1.query@x("select f1,if(f2='',null,f2) as f2,f3,f4 from test001 order by 1") |
3 |
=connect@l("oracle12c") |
4 |
=A3.query@x("select * from test001 order by 1") |
5 |
=join@fm(A2:t1,f1;A4:t2,f1) |
都按f1有序，A5的join用归并法，这时比较数值，不再是hash值。
顺利full join后就可以知道记录的新增、删除、修改情况了，代码如下：
A |
|
6 |
=A5.select(!t1).(t2) |
7 |
=A5.select(!t2).(t1) |
8 |
=A5.select(t1 && t2 && (${A2.fname().("t1."/~/"!=t2."/~).concat("||")})) |
A6找出t1没有但t2有的记录
A7找出t2没有但t1有的记录
A8找出t1、t2主键相等，但是其他字段中有不同的记录
无主键
当需要比对的数据没有明确主键时，需要全列数据比对：
A |
|
1 |
=connect@l("mysql8") |
2 |
=A1.query("SELECT COLUMN_NAME, DATA_TYPE FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME ='test001'") |
3 |
=A2.(if(#2=="varchar","if("/#1/"='',null,"/#1/") as"/#1,#1)).concat@c() |
4 |
=A1.query@x("select"/A3/"from test001 order by"/A2.(#).concat@c()) |
5 |
=connect@l("oracle12c") |
6 |
=A5.query("SELECT COLUMN_NAME, DATA_TYPE FROM USER_TAB_COLUMNS WHERE TABLE_NAME ='TEST001'") |
7 |
=A5.query@x("select"/A6.(#1).concat@c()/"from test001 order by"/A6.(#).concat@c()) |
8 |
=join@mf(A4:t1,${A4.fname().concat@c()};A7:t2,${A7.fname().concat@c()}) |
9 |
=A8.select(!t1).(t2) |
10 |
=A8.select(!t2).(t1) |
因为没有主键，所以只能找出两边任意字段可能无法对上的数据了
A6找出t1没有但t2有的记录
A7找出t2没有但t1有的记录
大数据量
当需要比对的数据量较大，无法一次性全部装入内存时
有主键
因为游标是一次性的，所以要一把找出增删改的记录，需要使用SPL的管道功能，代码如下：
A |
|
1 |
=connect@l("mysql8") |
2 |
=A1.query("SELECT COLUMN_NAME, DATA_TYPE FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME ='test001'") |
3 |
=A1.cursor@x("select f1,if(f2='',null,f2) as f2,f3,f4 from test001 order by 1") |
4 |
=connect@l("oracle12c") |
5 |
=A4.cursor@x("select * from test001 order by 1") |
6 |
=joinx@f(A3:t1,f1;A5:t2,f1) |
7 |
=channel(A6) |
8 |
=channel(A6) |
9 |
=channel(A6) |
10 |
>A7.select(!t1).(t2) |
11 |
>A7.fetch() |
12 |
>A8.select(!t2).(t1) |
13 |
>A8.fetch() |
14 |
>A9.select(t1 && t2 && (${A2.(#1).("t1."/~/"!=t2."/~).concat("||")})) |
15 |
>A9.fetch() |
16 |
=A6.skip() |
17 |
=A7.result() |
18 |
=A8.result() |
19 |
=A9.result() |
相比小数据量（序表），joinx默认要求游标对关联字段有序，不会采用hash，所以不存在数据类型不同导致结果不对的情况。类似的还有cs.group，也是默认要求对分组字段有序，不再hash比较。
A17找出t1没有但t2有的记录
A18找出t2没有但t1有的记录
A19找出t1、t2主键相等，但是其他字段中有不同的记录
无主键
同样利用管道，代码如下：
A |
|
1 |
=connect@l("mysql8") |
2 |
=A1.query("SELECT COLUMN_NAME, DATA_TYPE FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME ='test001'") |
3 |
=A2.(if(#2=="varchar","if("/#1/"='',null,"/#1/") as"/#1,#1)).concat@c() |
4 |
=A1.cursor@x("select"/A3/"from test001 order by"/A2.(#).concat@c()) |
5 |
=connect@l("oracle12c") |
6 |
=A5.query("SELECT COLUMN_NAME, DATA_TYPE FROM USER_TAB_COLUMNS WHERE TABLE_NAME ='TEST001'") |
7 |
=A5.cursor@x("select"/A6.(#1).concat@c()/"from test001 order by"/A6.(#).concat@c()) |
8 |
=joinx@f(A4:t1,${A2.(#1).concat@c()};A7:t2,${A2.(#1).concat@c()}) |
9 |
=channel(A8) |
10 |
=channel(A8) |
11 |
>A9.select(!t1).(t2) |
12 |
>A9.fetch() |
13 |
>A10.select(!t2).(t1) |
14 |
>A10.fetch() |
15 |
=A8.skip() |
16 |
=A9.result() |
17 |
=A10.result() |
A16找出t1没有但t2有的记录
A17找出t2没有但t1有的记录
比对结果的导出
ch.result会返回序表，当比对出的结果也很大时，为了避免内存不够用，可以将结果导出到文件，如下：
A |
|
1 |
=connect@l("mysql8") |
2 |
=A1.query("SELECT COLUMN_NAME, DATA_TYPE FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME ='test001'") |
3 |
=A2.(if(#2=="varchar","if("/#1/"='',null,"/#1/") as"/#1,#1)).concat@c() |
4 |
=A1.cursor@x("select"/A3/"from test001 order by"/A2.(#).concat@c()) |
5 |
=connect@l("oracle12c") |
6 |
=A5.query("SELECT COLUMN_NAME, DATA_TYPE FROM USER_TAB_COLUMNS WHERE TABLE_NAME ='TEST001'") |
7 |
=A5.cursor@x("select"/A6.(#1).concat@c()/"from test001 order by"/A6.(#).concat@c()) |
8 |
=joinx@f(A4:t1,${A2.(#1).concat@c()};A7:t2,${A2.(#1).concat@c()}) |
9 |
=channel(A8) |
10 |
=channel(A8) |
11 |
>A9.select(!t1).(t2) |
12 |
=A9.fetch(file("t2.b")) |
13 |
>A10.select(!t2).(t1) |
14 |
>A10.fetch(file("t1.b")) |
15 |
=A8.skip() |
A12和A14的ch.fetch(f)，f为集文件
英文版