《Debezium系列》Debezium-SqlServer的配置踩坑

冷不防 2022-11-29 13:05 595阅读 0赞

之前有过关于其他debezium数据库的配置，现在带来sqlServer的配置和踩坑。

话不多说，整起。。。。。

首先，按照惯例，先官方地址（毕竟有了官方才有了我这个。。。。）

官方传送门：[debezium][]

**环境**：

kafka2.4         
    zookeeper3.5.6          
    sqlserver2017standard

## 1.数据库配置 ##

需要为监控的数据库开启cdc功能以便debezium连接器能够获取日志文件。  
例如：

USE MyDB
    GO
    EXEC sys.sp_cdc_enable_db  #数据库开启cdc功能，注意不能为master数据库启动该功能
    GO

找出当前filegroup\_name的值，一般为PRIMARY

USE AdventureWorks;
    GO
    SELECT FILEGROUP_NAME(1) AS [Filegroup Name];
    GO

下一步为监控的表开启cdc

USE MyDB
    GO
    
    EXEC sys.sp_cdc_enable_table
    @source_schema = N'dbo',
    @source_name   = N'MyTable',       #填写要开启的表名
    @role_name     = N'MyRole',
    @filegroup_name = N'PRIMARY',      #文件组标识 (ID) 号的文件组名，可不写
    @supports_net_changes = 1
    GO
    
    #如果不知道该用户是否有权限访问这些cdc表，可以使用下列命令查看验证
    
    EXEC sys.sp_cdc_help_change_data_capture
    GO
    
    #如果为空的话，需要确保用户有相关访问权限，如果不为空则显示表的相关信息

***注意**：主节点上配置并启用了更改数据捕获。SQL Server不直接在副本上支持CDC。*

## 2.连接器配置 ##

下面为示例：

{
        "name": "sqlserver-connector",
        "config": {
            "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector", 
            "database.hostname": "xxx.xxx.xxx.xxx", 
            "database.port": "1433", 
            "database.user": "sa", 
            "database.password": "qwer*1234", 
            "database.dbname": "MyDB", 
            "database.server.name": "fullfillment", 
            "table.whitelist": "dbo.products", 
            "database.history.kafka.bootstrap.servers": "k1:9092,k2:9092,k3:9092", 
            "database.history.kafka.topic": "dbhistory.fullfillment" 
        }
    }

连接器监控数据库参数：  
![sqlServer参数相关配置][sqlServer]

**注意** ：  
关于database.history.kafka.topic这个参数，官方给出解释：

> The name of the database history topic where the connector will write and recover DDL statements. This topic is for internal use only and should not be used by consumers.

大意为会创建这么一个topic用于存放数据库的ddl相关语句用于记录，是内部使用的。消费者不要瞎去消费这个topic。

*  提交任务后会自动创建以下topic

fullfillment.dbo.customers
    fullfillment.dbo.products
    schema-changes.inventory

## 3. 实际测试 ##

1.创建连接器

curl -i -X POST "Accept:application/json" -H "Content-Type:application/json" k2:8083/connectors/ -d '{"name": "sqlserver-connector","config": {"connector.class":"io.debezium.connector.sqlserver.SqlServerConnector", "tasks.max" : "1", "database.hostname": "xxx.xxx.xxx.xxx", "database.port": "1433", "database.user": "sa", "database.password": "qwer*1234", "database.dbname": "testDB", "database.server.name": "fullfillment", "database.history.kafka.bootstrap.servers": "k1:9092,k2:9092,k3:9092", "database.history.kafka.topic": "schema-changes.inventory" }}'

2.t提交之后查看topic，相关topic可以选择自动创建，也可以自己创建  
例如下面中的customers，只有连接器捕获到了该监控表的变化，才会自动创建该topic  
![topic][]  
3.自动创建的topic命名形式为  
![topic names][]  
4.消费测试，进行消费发现成功  
![consumer test][]  
至此，流程走完了，连接器大成功啊。谢谢大家（鞠躬）。

## 4. 我们那些年遇到的错误 ##

1.  创建connector时出现Exception  
    ![exception][]
2.  The problem was that i don’t have active the service of Sql server Agent, after that i active, the CDC began to tracking the operations on the table.

看这里是没有开启SQL server的agent服务，这里进行开启。  
![enable agent][]  
3. 关于其中的行更新问题，即表结构变更说明：  
当更新行的主键/唯一键的列时，该行的键的值已更改，因此Debezium将输出三个事件：一个DELETE事件和一个具有行旧键的逻辑删除事件，然后是一个INSERT具有新键的事件排。  
Debezium能够捕获随时间变化的架构。由于在SQL Server中实现CDC的方式，有必要与数据库操作员合作以确保连接器在更新架构时继续产生数据更改事件。  
如前所述，Debezium使用SQL Server的更改数据捕获功能。这意味着SQL Server创建一个捕获表，其中包含在源表上执行的所有更改。不幸的是，捕获表是静态的，并且在源表结构更改时需要更新。此更新不是由连接器本身完成的，而是必须由具有提升特权的操作员执行。  
为源表启用CDC时，不支持所有架构更改。确定的此类异常之一是重命名列或更改其类型，SQL Server将不允许执行该操作。  
（上述翻译自google）

1.  热部署和冷部署  
    在进行字段增加减少时（alter table操作）：  
    分为两种部署方式：热部署和冷部署

*  冷部署：

> 1.  挂起生成数据库记录的应用程序
> 2.  等待Debezium流所有未流的更改
> 3.  停止连接器
> 4.  将所有更改应用于源表架构
> 5.  使用sys.sp\_cdc\_enable\_table具有唯一参数值的过程为更新源表创建新的捕获表@capture\_instance
> 6.  恢复应用
> 7.  启动连接器
> 8.  当Debezium从新的捕获表开始流式传输时，可以使用sys.sp\_cdc\_disable\_table参数@capture\_instance设置为旧捕获实例名称的存储过程删除旧的

*  热部署：

> 1.  将所有更改应用于源表架构
> 2.  使用sys.sp\_cdc\_enable\_table具有唯一参数值的过程为更新源表创建新的捕获表@capture\_instance
> 3.  当Debezium从新的捕获表开始流式传输时，可以使用sys.sp\_cdc\_disable\_table参数@capture\_instance设置为旧捕获实例名称的存储过程删除旧的

热模式更新有一个缺点。从数据库架构更新到创建新的捕获实例之间有一段时间。在此期间将要到达的所有更改都由具有旧结构的旧实例捕获。例如，这意味着对于新添加的列，在此期间产生的任何更改事件都将不包含该新列的字段。如果您的应用程序不允许这样的过渡期，我们建议您遵循冷模式更新。(**这也是官网原话**)

例子：

-- Modify the source table schema
    ALTER TABLE customers ADD phone_number VARCHAR(32);
    
    -- Create the new capture instance
    EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'customers', @role_name = NULL, @supports_net_changes = 0, @capture_instance = 'dbo_customers_v2';
    GO
    
    -- Insert new data
    INSERT INTO customers(first_name,last_name,email,phone_number) VALUES ('John','Doe','john.doe@example.com', '+1-555-123456');
    GO

kafka日志：

connect_1    | 2019-01-17 10:11:14,924 INFO   ||  Multiple capture instances present for the same table: Capture instance "dbo_customers" [sourceTableId=testDB.dbo.customers, changeTableId=testDB.cdc.dbo_customers_CT, startLsn=00000024:00000d98:0036, changeTableObjectId=1525580473, stopLsn=00000025:00000ef8:0048] and Capture instance "dbo_customers_v2" [sourceTableId=testDB.dbo.customers, changeTableId=testDB.cdc.dbo_customers_v2_CT, startLsn=00000025:00000ef8:0048, changeTableObjectId=1749581271, stopLsn=NULL]   [io.debezium.connector.sqlserver.SqlServerStreamingChangeEventSource]
    connect_1    | 2019-01-17 10:11:14,924 INFO   ||  Schema will be changed for ChangeTable [captureInstance=dbo_customers_v2, sourceTableId=testDB.dbo.customers, changeTableId=testDB.cdc.dbo_customers_v2_CT, startLsn=00000025:00000ef8:0048, changeTableObjectId=1749581271, stopLsn=NULL]   [io.debezium.connector.sqlserver.SqlServerStreamingChangeEventSource]
    ...
    connect_1    | 2019-01-17 10:11:33,719 INFO   ||  Migrating schema to ChangeTable [captureInstance=dbo_customers_v2, sourceTableId=testDB.dbo.customers, changeTableId=testDB.cdc.dbo_customers_v2_CT, startLsn=00000025:00000ef8:0048, changeTableObjectId=1749581271, stopLsn=NULL]   [io.debezium.connector.sqlserver.SqlServerStreamingChangeEventSource]

topic中：

...
         {
            "type": "string",
            "optional": true,
            "field": "phone_number"
         }
    ...
        "after": {
          "id": 1005,
          "first_name": "John",
          "last_name": "Doe",
          "email": "john.doe@example.com",
          "phone_number": "+1-555-123456"
        },

再重新部署之后，记得删除旧的capture instance

-- Drop the old capture instance
    EXEC sys.sp_cdc_disable_table @source_schema = 'dbo', @source_name = 'dbo_customers', @capture_instance = 'dbo_customers';
    GO

## 5. 小结 ##

总的来看，部署非常简单，对于其他数据库并没有什么复杂的配置，只需要开启CDC和配置相关权限即可。

此处对于该插件配置使用没有太多总结，看源码去咯。

## 重大问题 ##

*  sqlserver2017版本使用agent服务时，会出现存储过程超过32层的错误，导致agent中capture任务直接挂掉。  
    ![就是这个任务][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70_pic_center]  
    解决方法也很简单，官网打个补丁就完事了（666）  
    ![没错就是这个补丁][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70_pic_center 1]  
    还有你的数据库版本如果在这个版本之后，那，当我这个错误没说。。。。

[debezium]: https://github.com/debezium/debezium-examples/tree/master/tutorial
[sqlServer]: /images/20221124/a059725af30d4976b9076032365adfff.png
[topic]: /images/20221124/5f30b4f24b10489eabe5cb8f3379d74b.png
[topic names]: /images/20221124/f0f61392f7ba47dc95df8a4781f66f43.png
[consumer test]: /images/20221124/a7783bfb39a5424ba3c62a7630b61e69.png
[exception]: /images/20221124/5e62c60bf8344e6a8b94765f586cd1ba.png
[enable agent]: /images/20221124/cd8298d79ed54ab9aead669d552cbab2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70_pic_center]: /images/20221124/9176392a79514b638a9f2de272a16c8b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221124/0213bf1a556c42dcae1bec6b22753a1a.png