在Kettle里使用时间戳实现变化数据捕获（CDC）

朱雀 2022-02-15 08:10 538阅读 0赞

转自：[https://blog.csdn.net/wzy0623/article/details/53885552][https_blog.csdn.net_wzy0623_article_details_53885552]

版权声明：本文为博主原创文章，未经博主允许不得转载。    https://blog.csdn.net/wzy0623/article/details/53885552  
1. 建立测试表，插入数据。  
use test;    
    
create table t\_color (    
    id int unsigned not null auto\_increment primary key,    
    color varchar(10),    
    create\_date datetime,    
    last\_update timestamp    
)  engine=myisam;    
    
insert into t\_color (color,create\_date) values('black',now()),('green',now()),('red',now()),('blue',now());    
select \* from t\_color;    
2. 建立参数表存储最后一次的抽取时间。  
use test;    
    
\-- 创建时间戳表    
create table cdc\_time (    
last\_load datetime,    
current\_load datetime) engine=myisam;    
    
\-- 初始化数据    
insert into cdc\_time values ('1971-01-01 00:00:01','1971-01-01 00:00:01');    
select \* from cdc\_time;    
3. 创建初始化时间戳转换

说明：

把current\_load时间设置成作业的开始时间。通过“获取系统信息”完成这一功能，在这个步骤里创建一个“系统日期（变）”类型的字段，字段名是sysdate。然后创建一个“插入/更新”步骤，把“获取系统信息”步骤和“插入/更新”步骤连接起来。在“插入/更新”步骤的“更新字段”部分里，用流里的字段“sysdate”去更新表里的字段“current\_load”。另外还要设置“用来查询的关键字”部分，把表的“current\_load”的条件设置为“is not null”即可。

4. 创建查询变化数据的转换

说明：  
从t\_color表里抽取数据的查询语句使用开始日期和结束日期，左边闭区间，右边开区间。查询条件类似下面的语句：  
(create\_date >= last\_load and create\_date < current\_load) or (last\_update >= last\_load and last\_update < current\_load)  
这里需要两个表输入步骤，一个用来从cdc\_time表中抽取时间，另一个从t\_color表中抽取需要的数据。另外再看查询条件，可以发现last\_load和current\_load分别出现两次。就是说在第一个表输入步骤中，这些时间值需要被抽取出来两次。

select     
    last\_load last1,    
    current\_load cur1,    
    last\_load last2,    
    current\_load cur2    
from cdc\_time;  
在t\_color表输入步骤里，选中“替换 sql 语句里的变量”，在“从步骤插入数据”下拉列表里选中上个表输入步骤。在select语句里写入下面的查询条件：

where (create\_date >= ? and create\_date < ?) or (last\_update >= ? and last\_update < ?)  
前一个步骤传来的参数将替换上面语句里的问号，第一个问号的值是last1，第二个问号的值是cur1，等等。  
通过比较create\_date和last\_update的值是否相等，可以判断出是新增的还是更改的数据。  
case when create\_date = last\_update then 'new' else 'changed' end as flagfield  
把变更数据输出到文本文件里。  
5. 创建更新参数表的转换

说明：  
如果转换中没有发生任何错误，要把current\_load字段里的值复制到last\_load字段里。如果转换中发生了错误，时间戳需要保持不变。把current\_load字段里的值复制到last\_load字段里需要“执行sql语句”步骤，脚本如下：

update cdc\_time set last\_load = current\_load;  
cdc\_time表里之所以要有两个字段，是因为在加载过程中，会有新的数据被插入或更新，为避免脏读或死锁的情况，最好给create和update时间戳设定一个上限条件，也就是这里的current\_load字段。  
6. 创建作业

7. 测试  
\-- 运行作业  
\-- 查看diff文件

\-- 查看cdc\_time表

mysql> select \* from cdc\_time;  
\+---------------------+---------------------+  
| last\_load       | current\_load      |  
\+---------------------+---------------------+  
| 2014-12-16 11:10:05 | 2014-12-16 11:10:05 |  
\+---------------------+---------------------+  
1 row in set (0.00 sec)

\-- 修改数据

delete from t\_color where id=3;  
update t\_color set color='Grey' where id=1;  
insert into t\_color (color,create\_date) values('Yellow',now());  
commit;  
\-- 运行作业  
\-- 查看diff文件

\-- 查看cdc\_time表

mysql> select \* from cdc\_time;  
\+---------------------+---------------------+  
| last\_load       | current\_load      |  
\+---------------------+---------------------+  
| 2014-12-16 11:16:02 | 2014-12-16 11:16:02 |  
\+---------------------+---------------------+  
1 row in set (0.00 sec)

8. 总结  
  基于源数据的CDC要求源数据里有相关的属性列，ETL过程可以利用这些属性列，来判断出哪些数据是增量数据。最常见的属性列有以下两种：  
时间戳：这种方法至少需要一个更新时间戳，但最好有两个时间戳：一个插入时间戳，记录数据行什么时候创建；一个更新时间戳，记录数据行什么时候最后一次更新。  
序列：大多数数据库都有自增序列。如果数据库表用到了这种序列，就可以很容易识别出新插入的数据。  
  这两种方法都需要一个额外的数据库表来存储上一次更新时间或上一次抽取的最后一个序列号。在实践中，一般是在一个独立的模式下或在数据缓冲区里创建这个参数表，不能在数据仓库里创建，更不能在数据集市里创建。基于时间戳和自增序列的方法是CDC最简单的实现方式，所以也是最常用的方法。但是它的缺点也是很明显的，主要如下：  
区分插入操作和更新操作：只有当源系统包含了插入时间戳和更新时间戳两个字段，才能区别插入和更新，否则无法区分。  
删除记录的操作：不能捕获到删除操作，除非是逻辑删除，即记录没有真的删除，只是做了逻辑上的标志。  
多次更新检测：如果在一次同步周期内，数据被更新了多次，只能同步最后一次更新操作，中间的更新操作都丢失了。  
实时能力：时间戳和基于序列的数据抽取一般适用于批量操作，不适合于实时场景下的数据加载。  
\---------------------   
作者：wzy0623   
来源：CSDN   
原文：https://blog.csdn.net/wzy0623/article/details/53885552   
版权声明：本文为博主原创文章，转载请附上博文链接！

[https_blog.csdn.net_wzy0623_article_details_53885552]: https://blog.csdn.net/wzy0623/article/details/53885552