FlinkX SqlServer CDC实时采集原理与使用丨直播回顾

数栈研习社 2021-03-30

1764

3月23日，袋鼠云数栈技术研发团队工程师李珞（花名：土豆）为大家直播分享《FlinkX SqlServer CDC实时采集原理与使用》。错过直播的朋友可以钉钉扫描文末的二维码，加入钉钉群回看直播，或者在b站搜索“袋鼠云”观看视频。

b站视频网址：

https://www.bilibili.com/video/BV1xV411e718

下面带大家来回顾下本次直播的内容，本次直播土豆大佬主要从以下几个方面来为大家介绍FlinkX SqlServer CDC。

SqlServer CDC的基本介绍

1、基础

SqlServer官方支持CDC功能，其基本原理是对于开启CDC的表，当表中发生插入、更新和删除操作时，会在SqlServer的事务日志中记录这些事件，并在跟踪的关联表中添加更改的相关信息，并且提供了相关的系统函数查询这些修改的相关信息。

SqlServer官方从SqlServer 2008版本开始支持CDC。

文档连接如下：

https://docs.microsoft.com/zh-cn/sql/relational-databases/track-changes/about-change-data-capture-sql-server?view=sql-server-ver15

变更数据捕获的更改数据源为 SQL Server 事务日志。在将插入、更新和删除应用于跟踪的源表时，将会在日志中添加说明这些更改的项。日志用作捕获进程的输入来源，它会读取日志，并在跟踪的表的关联更改表中添加有关更改的信息。系统将提供一些函数，以枚举在更改表中指定范围内发生的更改，并以筛选的结果集的形式返回该值。通常，应用程序进程使用筛选的结果集在某种外部环境中更新源表示形式。

2、SqlServer代理

QL Server Agent代理服务，是sql server的一个标准服务，作用是代理执行所有sql的自动化任务，以及数据库事务性复制等无人值守任务。这个服务在默认安装情况下是停止状态，需要手动启动，或改为自动运动，否则sql的自动化任务都不会执行的，还要注意服务的启动帐户。简单的说就是启动了这个服务，捕获进程才会处理事务日志并将条目写入CDC表。

具体文档链接参考如下：

https://docs.microsoft.com/zh-cn/sql/ssms/agent/sql-server-agent?view=sql-server-ver15

3、SqlServer数据源配置CDC

查询SqlServer数据库版本
查询当前用户权限
查询数据库是否已经启用CDC(变更数据捕获)功能
对数据库数据库启用CDC(变更数据捕获)功能
查询表是否已经启用CDC(变更数据捕获)功能
对表启用CDC(变更数据捕获)功能
确认SQL Server 代理是否正常启动

4、原理
1）新增表

captured_column
change_tables
ddl_history
index_columns
lsn_time_mapping
dbo_${table_name}_CT

2）新增函数

fn_cdc_get_all_changes_
fn_cdc_get_net_changes_

5、采集原理
1）insert/delete
对于insert和delete类型的数据变更，对于每一行变更都会在对应的${schema}_${table}_CT表中增加一行记录。
对于insert，id，user_id，name记录的是insert之后的value值；
对于delete，id，user_id，name记录的是delete之前的value值；

2）update
a、更新了主键
此时，SqlServer数据库的做法是在同一事物内，先将原来的记录删除，然后再重新插入。
执行如下SQL，日志表如图所示：
UPDATE [dbo].[kudu] SET [id] = 2, [user_id] = '2', [name] = 'b' WHERE [id] = 1;

b、未更新主键
此时，SqlServer数据库的做法是直接更新字段信息。
执行如下SQL，日志表如图所示：
UPDATE [dbo].[kudu] SET [user_id] = '3', [name] = 'c' WHERE [id] = 2;

6、流程图

对于FlinkX SqlServer CDC实时采集插件，其基本原理便是以轮询的方式，循环调用fn_cdc_get_all_changes_函数，获取上次结束时的lsn与当前数据库最大lsn值之间的数据。

对于insert/delete类型的数据获取并解析一行，对于update类型获取并解析两行。解析完成后把数据传递到下游并记录当前解析到的数据的lsn，为下次轮询做准备。