空白诗007

这个屌丝很懒，什么也没留下！

热门标签

变更数据捕获

作者：空白诗007 | 2024-07-14 22:10:34

踩

变更数据捕获

变更数据捕获配置如下：

a. 在数据库上启用变更数据捕获（调用系统存储过程sys.sp_cdc_enable_db）；

b. 通过系统存储过程sys.sp_cdc_add_job创建捕获和清理Job（可选，如果没有捕获和清理Job，会在创建数据库中的第一个变更数据捕获时自动建立，自动建立的Job可以通过调用系统存储过程sys.sp_cdc_change_job来调整捕获和清理相关的一些选项）；

c. 在需要捕获变更数据的每个表上建立变更数据捕获实例（每个表上可以建立<=2个捕获实例，创建捕获实例使用系统存储过程sys.sp_cdc_enable_table）。

捕获的变更数据的查询包括：

a. sys.fn_cdc_get_min_lsn

返回指定捕获实例的有效性间隔的低端点（start_lsn）；

b. sys.fn_cdc_get_max_lsn

返回cdc.lsn_time_mapping系统表的最大日志序列号(LSN)；

c. cdc.fn_cdc_get_all_changes_<捕获实例>

针对在指定日志序列号(LSN)范围内应用到源表的每项更改均返回一行。如果源行在该间隔内有多项更改，则每项更改都会表示在返回的结果集中。除了返回更改数据外，四个元数据列还提供了将更改应用到另一个数据源所需的信息。行筛选选项可控制元数据列的内容以及结果集中返回的行。当指定“all”行筛选选项时，针对每项更改将只有一行来标识该更改。当指定“all update old”选项时，更新操作会表示为两行：一行包含更新之前已捕获列的值，另一行包含更新之后已捕获列的值。

此枚举函数是在对源表启用变更数据捕获时创建的。此函数名称是派生的，采用cdc.fn_cdc_get_all_changes_capture_instance格式，其中capture_instance是在对源表启用变更数据捕获时为捕获实例指定的值；

d. cdc.fn_cdc_get_net_changes_<capture_instance>

针对指定日志序列号(LSN)范围内每个已更改的源行返回一个净更改行。净更改行指：如果在LSN范围内源行具有多项更改，则该函数将返回反映该行最终内容的单一行。例如，如果事务在源表中插入一行，并且LSN范围内的后续事务更新了该行中的一个或多个列，则该函数将只返回一行，其中包含多个更新的列值。

此枚举函数是在对某源表启用变更数据捕获并指定净跟踪时创建的。函数名称是派生的，采用cdc.fn_cdc_get_net_changes_capture_instance格式，其中capture_instance是对变更数据捕获启用源表时为捕获实例指定的值；

e. sys.fn_cdc_map_time_to_lsn

为指定的时间返回cdc.lsn_time_mapping系统表中start_lsn列中的日志序列号(LSN)值；

f. sys.fn_cdc_has_column_changed

标识指定的更新掩码是否指示已更新关联的更改行中的指定列。

下面的T-SQL示例创建一个测试数据库，并在测试数据库中演示配置变更数据捕获及查询捕获结果。

-- ====================================================

-- 测试的数据库

USE master;

CREATE DATABASE DB_test;

-- 启用变更数据捕获

USE DB_test;

EXEC sys.sp_cdc_enable_db;

-- ====================================================

-- 检查SQL Server Agent 服务的状态，如果未启动，则启动它

DECLARE

@agnt_service sysname;

SET @agnt_service = N'SQLServerAgent';

DECLARE @tb_agent_status TABLE(

state varchar(50)

);

INSERT @tb_agent_status

EXEC master.sys.xp_servicecontrol

N'QUERYSTATE',

@agnt_service;

IF NOT EXISTS(

SELECT * FROM @tb_agent_status

WHERE state = N'Running.')

EXEC master.sys.xp_servicecontrol

N'START',

@agnt_service;

-- ====================================================

-- 测试的表

USE DB_test;

CREATE TABLE dbo.tb(

id int

CONSTRAINT PK_tb_id PRIMARY KEY,

col1 int,

col2 varchar(10),

col3 nvarchar(max),

col4 varbinary(max),

col5 xml

);

-- 创建一个变更数据捕获实例- 所有列

-- 创建数据库中的第一个变更数据捕获实例的时候，数据捕获和清理的JOB 会自动创建

-- 可以通过sys.sp_cdc_change_job 这个存储过程去调整捕获和清理的相关设置

-- 也可以在创建第一个变更数据捕获实例前，使用sys.sp_cdc_add_job去创建数据捕获和清理Job，在创建时做好相关的设置

EXEC sys.sp_cdc_enable_table

@source_schema = N'dbo',

@source_name = N'tb',

@capture_instance = N'dbo_tb',

@role_name = NULL;

-- 创建一个变更数据捕获实例- 特定列

EXEC sys.sp_cdc_enable_table

@source_schema = N'dbo',

@source_name = N'tb',

@capture_instance = N'dbo_tb_col',

@role_name = NULL,

@captured_column_list = N'id,col1,col2';

-- ====================================================

-- 数据测试

-- a. 插入初始数据

INSERT dbo.tb(

id,

col1, col2, col3, col4, col5)

VALUES(

1, 'AA', 'AAA', 0x1, '<a>aa</a>'),

(

2, 'BB', 'BBB', 0x2, '<b/>'),

(

3, 'CC', 'CCC', 0x2, '<c/>');

WITH

LSN AS(

SELECT

from_lsn = sys.fn_cdc_get_min_lsn(N'dbo_tb'),

to_lsn = sys.fn_cdc_get_max_lsn()

CHG_ALL AS(

SELECT

CHG.*

FROM LSN

CROSS APPLY cdc.fn_cdc_get_all_changes_dbo_tb(LSN.from_lsn, LSN.to_lsn, 'ALL UPDATE OLD') CHG

CHG_NET AS(

SELECT

CHG.*

FROM LSN

CROSS APPLY cdc.fn_cdc_get_net_changes_dbo_tb(LSN.from_lsn, LSN.to_lsn, 'ALL') CHG

)

SELECT * FROM CHG_ALL;

-- b. 更新数据

BEGIN TRAN;

UPDATE dbo.tb SET

col1 = 11

WHERE id = 1;

UPDATE dbo.tb SET

col1 = 111

WHERE id = 1;

COMMIT TRAN;

WITH

LSN AS(

SELECT

from_lsn = sys.fn_cdc_get_min_lsn(N'dbo_tb'),

to_lsn = sys.fn_cdc_get_max_lsn()

CHG_ALL AS(

SELECT

CHG.*

FROM LSN

CROSS APPLY cdc.fn_cdc_get_all_changes_dbo_tb(LSN.from_lsn, LSN.to_lsn, 'ALL UPDATE OLD') CHG

CHG_NET AS(

SELECT

CHG.*

FROM LSN

CROSS APPLY cdc.fn_cdc_get_net_changes_dbo_tb(LSN.from_lsn, LSN.to_lsn, 'ALL') CHG

)

SELECT * FROM CHG_ALL;

-- c. 更新xml 和varbinary(max) 数据

UPDATE dbo.tb SET

col5.modify('replace value of /a[1]/text()[1] with "replace"')

WHERE id = 1;

UPDATE dbo.tb SET

col5.modify('insert <a>1</a> as last into /')

WHERE id = 2;

UPDATE dbo.tb SET

col4 = col4 + 0x12345

WHERE id = 3;

WITH

LSN AS(

SELECT

from_lsn = sys.fn_cdc_get_min_lsn(N'dbo_tb'),

to_lsn = sys.fn_cdc_get_max_lsn()

CHG_ALL AS(

SELECT

CHG.*

FROM LSN

CROSS APPLY cdc.fn_cdc_get_all_changes_dbo_tb(LSN.from_lsn, LSN.to_lsn, 'ALL UPDATE OLD') CHG

CHG_NET AS(

SELECT

CHG.*

FROM LSN

CROSS APPLY cdc.fn_cdc_get_net_changes_dbo_tb(LSN.from_lsn, LSN.to_lsn, 'ALL') CHG

)

SELECT * FROM CHG_ALL;

-- d. 更新主键

UPDATE dbo.tb SET

id = 11

WHERE id = 1;

INSERT dbo.tb(

id,

col1, col2, col3, col4, col5)

VALUES(

1, 'AA', 'AAA', 0x1, '<a>aa</a>');

WITH

LSN AS(

SELECT

from_lsn = sys.fn_cdc_get_min_lsn(N'dbo_tb'),

to_lsn = sys.fn_cdc_get_max_lsn()

CHG_ALL AS(

SELECT

CHG.*

FROM LSN

CROSS APPLY cdc.fn_cdc_get_all_changes_dbo_tb(LSN.from_lsn, LSN.to_lsn, 'ALL UPDATE OLD') CHG

CHG_NET AS(

SELECT

CHG.*

FROM LSN

CROSS APPLY cdc.fn_cdc_get_net_changes_dbo_tb(LSN.from_lsn, LSN.to_lsn, 'ALL') CHG

)

SELECT * FROM CHG_ALL;

-- ====================================================

-- 删除测试

/*--

USE master;

ALTER DATABASE DB_test SET

SINGLE_USER

WITH

ROLLBACK AFTER 0;

DROP DATABASE DB_test;

--*/

声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop】