跳到主要内容
版本:3.x

SQL Server

安装 Agent 后,您需要在 Tapdata Cloud 平台为 Agent 和 SQL Server 数据库建立连接,完成操作后即可在数据复制/开发任务中使用该数据源。本文介绍建立连接前的准备工作(如授权账号等)。

支持版本

SQL Server 2005、2008、2008 R2、2012、2014、2016、2017

作为源库

提示

由于 CDC 支持从 SQLServer 2008 开始支持,对于较早的版本,您需要使用 Custom SQL 功能来模拟更改数据捕获,在从旧版本复制数据时,需要考虑以下几点:

  • 源表必须有一个更改跟踪列,比如 LAST_UPDATED_TIME,它在每次插入或更新记录时都会更新。
  • 创建数据同步任务时,任务的同步类型选择为全量,将重复运行自定义 SQL设置为 True,同时在映射设计上提供适当的自定义 SQL。
  1. 以 sysadmin 的身份登录到 SQLServer Management Studio 或 sqlcmd。

  2. 查找 mssql-conf 工具并开启代理服务。

    mssql-conf set sqlagent.enabled true
  3. 执行下述命令,启用数据库、数据表的增量复制。

--启用增量复制
use [数据库名称]
go
EXEC sys.sp_cdc_enable_db
go

--查看是否启用增量复制
SELECT [name], database_id, is_cdc_enabled
FROM sys.databases
WHERE [name] = N'[数据库名称]'
go
  1. 如果对增量同步表的字段执行了 DDL 操作(如增加字段),您需要执行下述操作重启 CDC,否则可能出现数据无法同步或报错等情况。

    --关闭该表的 CDC
    go
    EXEC sys.sp_cdc_disable_table
    @source_schema = N'[Schema]',
    @source_name = N'[Table]',
    @capture_instance = N'[Schema_Table]'
    go
    // capture_instance一般为schema_table的格式拼接而成,可以通过以下命令,查询实际的值
    exec sys.sp_cdc_help_change_data_capture
    @source_schema = N'[Schema]',
    @source_name = N'[Table]';


    --启动该表的 CDC
    use [数据库名称]
    go
    EXEC sys.sp_cdc_enable_table
    @source_schema = N'[Schema]',
    @source_name = N'[Table]',
    @role_name = N'[Role]'
    go
  2. (可选)如需向从节点读取增量数据以实现数据同步,您需要为从节点设置上述步骤。

  3. 创建用于数据同步/开发任务的账号并授予 sysadmin 权限,具体操作,见 CREATE USER

下一步

连接 SQL Server 数据库

扩展阅读

  • 如遇到本文未覆盖到的问题,请参考微软官方文档

  • 清理 CDC 日志

    SQL Server 不会自动清理增量数据日志,需要进行如下设置开启清理任务。

    --retention 的单位为分钟,本处设定清理周期为2天
    USE AdventureWorks2012;
    GO
    EXECUTE sys.sp_cdc_change_job
    @job_type = N'cleanup',
    @retention = 2880;
    GO
  • 开启全库 CDC

    -- 全局替换 将 TAPDATA 替换为实际的数据库名
    -- 全局替换 将 INSURANCE 替换为实际的 schema 名称
    USE TAPDATA
    GO
    EXEC sys.sp_cdc_enable_db
    GO

    declare @table_name varchar(100)
    declare @database_name varchar(100)
    declare @schema_name varchar(100)

    set @database_name = 'TAPDATA'
    set @schema_name = 'INSURANCE'

    declare my_cursor cursor for SELECT TABLE_NAME
    FROM TAPDATA.INFORMATION_SCHEMA.TABLES
    where TABLE_CATALOG = @database_name
    and TABLE_SCHEMA = @schema_name;
    open my_cursor
    fetch next from my_cursor into @table_name
    while @@FETCH_STATUS = 0
    begin
    begin try
    exec sys.sp_cdc_enable_table
    @source_schema = @schema_name,
    @source_name = @table_name,
    @role_name = NULL
    end try
    begin catch
    print('[ERROR] ' + @table_name)
    end catch

    fetch next from my_cursor into @table_name
    end
    close my_cursor
    deallocate my_cursor
  • 关闭全局 CDC

    -- 全局替换 将 TAPDATA 替换为实际的数据库名
    -- 全局替换 将 INSURANCE 替换为实际的 schema 名称
    USE TAPDATA
    GO

    declare @table_name varchar(100)
    declare @database_name varchar(100)
    declare @schema_name varchar(100)

    set @database_name = 'TAPDATA'
    set @schema_name = 'INSURANCE'

    declare my_cursor cursor for SELECT TABLE_NAME
    FROM TAPDATA.INFORMATION_SCHEMA.TABLES
    where TABLE_CATALOG = @database_name
    and TABLE_SCHEMA = @schema_name;
    open my_cursor
    fetch next from my_cursor into @table_name
    while @@FETCH_STATUS = 0
    begin
    begin try
    EXEC sys.sp_cdc_disable_table
    @source_schema = @schema_name,
    @source_name = @table_name,
    @capture_instance = 'all';
    end try
    begin catch
    print ('[ERROR] ' + @table_name)
    end catch

    fetch next from my_cursor into @table_name
    end
    close my_cursor
    deallocate my_cursor

    EXEC sys.sp_cdc_disable_db
    GO