Debezium SQL Server 连接器第一次连接到 SQL Server 数据库或集群时，它会获取数据库中模式的一致快照。初始快照完成后，连接器会持续捕获提交到启用了 CDC 的 SQL Server 数据库的 INSERT、UPDATE 或 DELETE 操作的行级更改。连接器为每个数据更改操作生成事件，并将它们流式传输到 Kafka 主题。连接器将表的所有事件流式传输到专用的 Kafka 主题。然后应用程序和服务可以使用来自该主题的数据更改事件记录。

一、Debezium概述

Debezium SQL Server 连接器基于 SQL Server 2016 Service Pack 1 (SP1) 和更高版本的标准版或企业版中提供的更改数据捕获功能。 SQL Server 捕获进程监视指定的数据库和表，并将更改存储到专门创建的具有存储过程外观的更改表中。

要启用 Debezium SQL Server 连接器以捕获数据库操作的更改事件记录，您必须首先在 SQL Server 数据库上启用更改数据捕获。必须在数据库和要捕获的每个表上都启用 CDC。在源数据库上设置 CDC 后，连接器可以捕获数据库中发生的行级 INSERT、UPDATE 和 DELETE 操作。连接器将每个源表的事件记录写入专门用于该表的 Kafka 主题。每个捕获的表都存在一个主题。客户端应用程序读取它们遵循的数据库表的 Kafka 主题，并且可以响应它们从这些主题中使用的行级事件。

连接器第一次连接到 SQL Server 数据库或集群时，它会为所有配置为捕获更改的表拍摄模式的一致快照，并将此状态流式传输到 Kafka。快照完成后，连接器会持续捕获随后发生的行级更改。通过首先建立所有数据的一致视图，连接器可以继续读取，而不会丢失在快照发生时所做的任何更改。

Debezium SQL Server 连接器可以容忍故障。当连接器读取更改并产生事件时，它会定期记录事件在数据库日志中的位置（LSN/日志序列号）。如果连接器因任何原因（包括通信故障、网络问题或崩溃）停止，则在重新启动后，连接器将从它读取的最后一点继续读取 SQL Server CDC 表。

注意：偏移量会定期提交。它们不会在更改事件发生时提交。因此，在中断之后，可能会生成重复事件。

容错也适用于快照。也就是说，如果连接器在快照期间停止，则连接器在重新启动时会开始新的快照。

二、SQL Server 连接器的工作原理

为了优化配置和运行 Debezium SQL Server 连接器，了解连接器如何执行快照、流式传输更改事件、确定 Kafka 主题名称以及使用元数据会很有帮助。

1.Snapshots

SQL Server CDC 并非旨在存储数据库更改的完整历史记录。为了让 Debezium SQL Server 连接器为数据库的当前状态建立基线，它使用了一个称为快照的过程。

您可以配置连接器创建快照的方式。默认情况下，连接器的快照模式设置为初始。基于此初始快照模式，连接器第一次启动时，会执行数据库的初始一致快照。此初始快照捕获与为连接器配置的包含和排除属性定义的条件匹配的任何表的结构和数据（例如，table.include.list、column.include.list、table.exclude.list、等等）。

当连接器创建快照时，它会完成以下任务：

• 1.确定要捕获的表。
• 2.获得对启用了 CDC 的 SQL Server 表的锁定，以防止在创建快照期间发生结构更改。锁的级别由 snapshot.isolation.mode 配置选项决定。
• 3.读取服务器事务日志中的最大日志序列号 (LSN) 位置。
• 4.捕获所有相关表的结构。
• 5.如有必要，释放在步骤 2 中获得的锁。在大多数情况下，锁只保留很短的时间。
• 6.根据步骤 3 中读取的 LSN 位置扫描要捕获的 SQL Server 源表和模式，为表中的每一行生成 READ 事件，并将事件写入表的 Kafka 主题。
• 7.在连接器偏移中记录快照的成功完成。

生成的初始快照捕获为 CDC 启用的表中每一行的当前状态。从这个基线状态，连接器会在后续更改发生时捕获它们。

Ad hoc snapshots

默认情况下，连接器仅在首次启动后才运行初始快照操作。在这个初始快照之后，在正常情况下，连接器不会重复快照过程。连接器捕获的任何未来更改事件数据仅通过流式处理进入。

但是，在某些情况下，连接器在初始快照期间获得的数据可能会变得陈旧、丢失或不完整。为了提供一种重新捕获表数据的机制，Debezium 包含一个执行临时快照的选项。数据库中的以下更改可能会导致执行临时快照：

• 修改连接器配置以捕获一组不同的表。
• Kafka 主题被删除，必须重建。
• 由于配置错误或其他问题而发生数据损坏。

您可以通过启动所谓的临时快照来为之前捕获快照的表重新运行快照。即席快照需要使用信令表。您可以通过向 Debezium 信号表发送信号请求来启动临时快照。

当您启动现有表的临时快照时，连接器会将内容附加到表已存在的主题中。如果删除了以前存在的主题，如果启用了自动主题创建，Debezium 可以自动创建主题。

即席快照信号指定要包含在快照中的表。快照可以捕获数据库的全部内容，或仅捕获数据库中表的子集。

您可以通过向信令表发送执行快照消息来指定要捕获的表。将执行快照信号的类型设置为增量，并提供要包含在快照中的表的名称，如下表所述：

表 1. 即席执行快照信号记录示例

Triggering an ad hoc snapshot

您可以通过将具有执行快照信号类型的条目添加到信令表来启动临时快照。连接器处理完消息后，将开始快照操作。快照进程读取第一个和最后一个主键值，并将这些值用作每个表的起点和终点。根据表中的条目数和配置的块大小，Debezium 将表划分为块，并继续对每个块进行快照，一次一个。

目前，执行快照操作类型仅触发增量快照。有关详细信息，请参阅增量快照。

Incremental snapshots

为了提供管理快照的灵活性，Debezium 包含一个补充快照机制，称为增量快照。增量快照依赖于 Debezium 机制将信号发送到 Debezium 连接器。

在增量快照中，Debezium 不是像在初始快照中那样一次捕获数据库的完整状态，而是在一系列可配置的块中分阶段捕获每个表。您可以指定您希望快照捕获的表以及每个块的大小。块大小决定了快照在数据库上的每次提取操作期间收集的行数。增量快照的默认块大小为 1 KB。

随着增量快照的进行，Debezium 使用水印来跟踪其进度，维护它捕获的每个表行的记录。与标准初始快照过程相比，这种分阶段捕获数据的方法具有以下优势：

• 您可以在流式数据捕获的同时运行增量快照，而不是将流式传输推迟到快照完成。连接器在整个快照过程中继续从更改日志中捕获近乎实时的事件，并且两个操作都不会阻塞另一个操作。
• 您可以在流式数据捕获的同时运行增量快照，而不是将流式传输推迟到快照完成。连接器在整个快照过程中继续从更改日志中捕获近乎实时的事件，并且两个操作都不会阻塞另一个操作。
• 您可以随时按需运行增量快照，并根据需要重复该过程以适应数据库更新。例如，您可以在修改连接器配置以将表添加到其 table.include.list 属性后重新运行快照。

Incremental snapshot process


当您运行增量快照时，Debezium 按主键对每个表进行排序，然后根据配置的块大小将表拆分为块。逐块工作，然后捕获块中的每个表行。对于它捕获的每一行，快照都会发出一个 READ 事件。该事件表示块的快照开始时行的值。

随着快照的进行，其他进程可能会继续访问数据库，可能会修改表记录。为了反映此类更改，INSERT、UPDATE 或 DELETE 操作将照常提交到事务日志。同样，正在进行的 Debezium 流式处理继续检测这些更改事件并将相应的更改事件记录发送到 Kafka。

Debezium 如何解决具有相同主键的记录之间的冲突


在某些情况下，流式处理发出的 UPDATE 或 DELETE 事件是乱序接收的。也就是说，流式处理可能会在快照捕获包含该行的 READ 事件的块之前发出一个修改表行的事件。当快照最终为该行发出相应的 READ 事件时，它的值已经被取代。为了确保以正确的逻辑顺序处理乱序到达的增量快照事件，Debezium 采用了一种缓冲方案来解决冲突。只有在解决了快照事件和流事件之间的冲突后，Debezium 才会向 Kafka 发出事件记录。

Snapshot window

为了帮助解决延迟到达的 READ 事件和修改同一表行的流事件之间的冲突，Debezium 采用了所谓的快照窗口。快照窗口划分了增量快照捕获指定表块数据的时间间隔。在一个块的快照窗口打开之前，Debezium 遵循其通常的行为并从事务日志直接向下游发送事件到目标 Kafka 主题。但是从特定块的快照打开的那一刻起，直到它关闭，Debezium 执行重复数据删除步骤以解决具有相同主键的事件之间的冲突。

对于每个数据集合，Debezium 发出两种类型的事件，并将它们的记录存储在单个目标 Kafka 主题中。它直接从表中捕获的快照记录作为 READ 操作发出。同时，随着用户不断更新数据集合中的记录，事务日志也会更新以反映每次提交，Debezium 会针对每次更改发出 UPDATE 或 DELETE 操作。

当快照窗口打开时，Debezium 开始处理快照块，它将快照记录传递到内存缓冲区。在快照窗口期间，缓冲区中 READ 事件的主键与传入流事件的主键进行比较。如果未找到匹配项，则将流式事件记录直接发送到 Kafka。如果 Debezium 检测到匹配，它会丢弃缓冲的 READ 事件，并将流式记录写入目标主题，因为流式事件在逻辑上取代了静态快照事件。块的快照窗口关闭后，缓冲区仅包含不存在相关事务日志事件的 READ 事件。 Debezium 将这些剩余的 READ 事件发送到表的 Kafka 主题。

连接器对每个快照块重复该过程。

Triggering an incremental snapshot

目前，启动增量快照的唯一方法是将临时快照信号发送到源数据库上的信令表。您将信号作为 SQL INSERT 查询提交到表。 Debezium 检测到信号表中的变化后，它会读取信号，并运行请求的快照操作。

您提交的查询指定要包含在快照中的表，并且可以选择指定快照操作的类型。目前，快照操作的唯一有效选项是默认值增量。

要指定要包含在快照中的表，请提供列出表的数据集合数组，例如，

{“数据集合”：[“public.MyFirstTable”，“public.MySecondTable”]}

增量快照信号的数据集合数组没有默认值。如果 data-collections 数组为空，Debezium 会检测到不需要任何操作并且不执行快照。

注意：

如果要包含在快照中的表的名称在数据库、模式或表的名称中包含点 (.)，则要将表添加到 data-collections 数组中，则必须转义双引号中的名称。

例如，要包含一个存在于公共架构中且名称为 My.Table 的表，请使用以下格式：“public”.“My.Table”。

先决条件

• 信令已启用。

  • 
    源数据库上存在信令数据集合，连接器配置为捕获它。
  • 
    信令数据集合在 signal.data.collection 属性中指定。

程序

发送 SQL 查询以将临时增量快照请求添加到信令表：

INSERT INTO _<signalTable>_ (id, type, data) VALUES (_'<id>'_, _'<snapshotType>'_, '{"data-collections": ["_<tableName>_","_<tableName>_"],"type":"_<snapshotType>_"}');

例如：

INSERT INTO myschema.debezium_signal (id, type, data) VALUES('ad-hoc-1', 'execute-snapshot', '{"data-collections": ["schema1.table1", "schema2.table2"],"type":"incremental"}');

2.读取变更数据表

当连接器第一次启动时，它会对捕获的表的结构进行结构快照，并将此信息保存到其内部数据库历史主题中。然后连接器为每个源表标识一个更改表，并完成以下步骤。

• 1.对于每个更改表，连接器读取在上次存储的最大 LSN 和当前最大 LSN 之间创建的所有更改。
• 2.连接器根据其提交 LSN 和更改 LSN 的值按升序对读取的更改进行排序。这种排序顺序可确保 Debezium 以与它们在数据库中发生的顺序相同的顺序重放更改。
• 3.连接器将提交和更改 LSN 作为偏移量传递给 Kafka Connect。
• 4.连接器存储最大 LSN 并从步骤 1 重新启动该过程。

重新启动后，连接器从它读取的最后一个偏移量（提交和更改 LSN）恢复处理。

连接器能够检测是否为包含的源表启用或禁用了 CDC，并调整其行为。

3.限制

SQL Server 特别要求基础对象是一个表，以便创建变更捕获实例。因此，SQL Server 不支持从索引视图（又名物化视图）捕获更改，因此 Debezium SQL Server 连接器也不支持。

4.topic名称

默认情况下，SQL Server 连接器将表中发生的所有 INSERT、UPDATE 和 DELETE 操作的事件写入特定于该表的单个 Apache Kafka 主题。连接器使用以下约定来命名更改事件主题：<serverName>.<schemaName>.<tableName>

以下列表提供了默认名称组件的定义：

服务器名称

• 服务器的逻辑名称，由 database.server.name 配置属性指定。

模式名称

• 发生更改事件的数据库模式的名称。

表名

• 发生更改事件的数据库表的名称。

例如，如果fulfillment 是服务器名称，dbo 是模式名称，并且数据库包含名称为 products、products_on_hand、customers 和 orders 的表，则连接器会将更改事件记录流式传输到以下 Kafka 主题：

• fulfillment.dbo.products
• fulfillment.dbo.products_on_hand
• fulfillment.dbo.customers
• fulfillment.dbo.orders

连接器应用类似的命名约定来标记其内部数据库历史主题、模式更改主题和事务元数据主题。

如果默认主题名称不符合您的要求，您可以配置自定义主题名称。要配置自定义主题名称，请在逻辑主题路由 SMT 中指定正则表达式。有关使用逻辑主题路由 SMT 自定义主题命名的更多信息，请参阅主题路由。

5.架构更改主题

对于启用了 CDC 的每个表，Debezium SQL Server 连接器存储应用于数据库中捕获的表的架构更改事件的历史记录。连接器将架构更改事件写入名为 <serverName> 的 Kafka 主题，其中 serverName 是在 database.server.name 配置属性中指定的逻辑服务器名称。

连接器发送到架构更改主题的消息包含有效负载，并且（可选）还包含更改事件消息的架构。架构更改事件消息的有效负载包括以下元素：

数据库名称：应用语句的数据库的名称。 databaseName 的值用作消息键。

表更改：架构更改后整个表架构的结构化表示。 tableChanges 字段包含一个数组，其中包含表中每一列的条目。由于结构化表示以 JSON 或 Avro 格式呈现数据，因此消费者可以轻松读取消息，而无需先通过 DDL 解析器对其进行处理。

重要：

当连接器被配置为捕获表时，它不仅将表模式更改的历史存储在模式更改主题中，还存储在内部数据库历史主题中。内部数据库历史主题仅供连接器使用，不适合消费应用程序直接使用。确保需要有关架构更改通知的应用程序仅使用来自架构更改主题的信息。

连接器向其模式更改主题发出的消息格式处于孵化状态，可以在不通知的情况下更改。

当以下事件发生时，Debezium 会向模式更改主题发出消息：

• 您为表启用 CDC。
• 您为表禁用 CDC。
• 您可以通过遵循模式演变过程来更改启用了 CDC 的表的结构。

示例：发送到 SQL Server 连接器架构更改主题的消息

以下示例显示了架构更改主题中的消息。该消息包含表模式的逻辑表示。

{
  "schema": {
  ...
  },
  "payload": {
    "source": {
      "version": "1.9.6.Final",
      "connector": "sqlserver",
      "name": "server1",
      "ts_ms": 1588252618953,
      "snapshot": "true",
      "db": "testDB",
      "schema": "dbo",
      "table": "customers",
      "change_lsn": null,
      "commit_lsn": "00000025:00000d98:00a2",
      "event_serial_no": null
    },
    "databaseName": "testDB", 
    "schemaName": "dbo",
    "ddl": null, 
    "tableChanges": [ 
      {
        "type": "CREATE", 
        "id": "\"testDB\".\"dbo\".\"customers\"", 
        "table": { 
          "defaultCharsetName": null,
          "primaryKeyColumnNames": [ 
            "id"
          ],
          "columns": [ 
            {
              "name": "id",
              "jdbcType": 4,
              "nativeType": null,
              "typeName": "int identity",
              "typeExpression": "int identity",
              "charsetName": null,
              "length": 10,
              "scale": 0,
              "position": 1,
              "optional": false,
              "autoIncremented": false,
              "generated": false
            },
            {
              "name": "first_name",
              "jdbcType": 12,
              "nativeType": null,
              "typeName": "varchar",
              "typeExpression": "varchar",
              "charsetName": null,
              "length": 255,
              "scale": null,
              "position": 2,
              "optional": false,
              "autoIncremented": false,
              "generated": false
            },
            {
              "name": "last_name",
              "jdbcType": 12,
              "nativeType": null,
              "typeName": "varchar",
              "typeExpression": "varchar",
              "charsetName": null,
              "length": 255,
              "scale": null,
              "position": 3,
              "optional": false,
              "autoIncremented": false,
              "generated": false
            },
            {
              "name": "email",
              "jdbcType": 12,
              "nativeType": null,
              "typeName": "varchar",
              "typeExpression": "varchar",
              "charsetName": null,
              "length": 255,
              "scale": null,
              "position": 4,
              "optional": false,
              "autoIncremented": false,
              "generated": false
            }
          ]
        }
      }
    ]
  }
}

表 3. 发送到模式更改主题的消息中的字段描述

在连接器发送到架构更改主题的消息中，键是包含架构更改的数据库的名称。在以下示例中，有效负载字段包含密钥：

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "string",
        "optional": false,
        "field": "databaseName"
      }
    ],
    "optional": false,
    "name": "io.debezium.connector.sqlserver.SchemaChangeKey"
  },
  "payload": {
    "databaseName": "testDB"
  }
}

三.数据更改事件

Debezium SQL Server 连接器为每个行级 INSERT、UPDATE 和 DELETE 操作生成数据更改事件。每个事件都包含一个键和一个值。键和值的结构取决于已更改的表。

Debezium 和 Kafka Connect 是围绕连续的事件消息流设计的。但是，这些事件的结构可能会随着时间的推移而发生变化，这对于消费者来说可能难以处理。为了解决这个问题，每个事件都包含其内容的模式，或者，如果您使用的是模式注册表，消费者可以使用它从注册表中获取模式的模式 ID。这使得每个事件都是独立的。

以下骨架 JSON 显示了更改事件的基本四个部分。但是，如何配置您选择在应用程序中使用的 Kafka Connect 转换器决定了这四个部分在更改事件中的表示。仅当您将转换器配置为生成它时，模式字段才处于更改事件中。同样，仅当您将转换器配置为生成它时，事件键和事件有效负载才在更改事件中。如果您使用 JSON 转换器并将其配置为生成所有四个基本更改事件部分，则更改事件具有以下结构：

{
 "schema": {    1
   ...
  },
 "payload": {   2
   ...
 },
 "schema": {    3
   ...
 },
 "payload": {   4
   ...
 },
}

表 4. 变更事件基本内容概览

1.更改事件键

更改事件的键包含更改表键的模式和更改行的实际键。在连接器创建事件时，模式及其相应的有效负载都包含已更改表的主键（或唯一键约束）中每一列的字段。

请考虑以下客户表，该表后面是该表的更改事件键的示例。

CREATE TABLE customers (
  id INTEGER IDENTITY(1001,1) NOT NULL PRIMARY KEY,
  first_name VARCHAR(255) NOT NULL,
  last_name VARCHAR(255) NOT NULL,
  email VARCHAR(255) NOT NULL UNIQUE
);

示例更改事件键

捕获对客户表的更改的每个更改事件都具有相同的事件键模式。只要客户表具有前面的定义，每个捕获客户表更改的更改事件都具有以下关键结构，在 JSON 中，如下所示：

{
    "schema": { 
        "type": "struct",
        "fields": [ 
            {
                "type": "int32",
                "optional": false,
                "field": "id"
            }
        ],
        "optional": false, 
        "name": "server1.dbo.customers.Key" 
    },
    "payload": { 
        "id": 1004
    }
}

表 5. 更改事件键的说明

注意：

尽管 column.exclude.list 和 column.include.list 连接器配置属性允许您仅捕获表列的子集，但主键或唯一键中的所有列始终包含在事件键中。

警告：

如果表没有主键或唯一键，则更改事件的键为空。这是有道理的，因为无法唯一标识没有主键或唯一键约束的表中的行。

2.更改事件值

更改事件中的值比键复杂一些。与键一样，值也有模式部分和有效负载部分。架构部分包含描述有效负载部分的信封结构的架构，包括其嵌套字段。创建、更新或删除数据的操作的更改事件都有一个带有信封结构的值负载。

考虑用于显示更改事件键示例的相同示例表：

CREATE TABLE customers (
  id INTEGER IDENTITY(1001,1) NOT NULL PRIMARY KEY,
  first_name VARCHAR(255) NOT NULL,
  last_name VARCHAR(255) NOT NULL,
  email VARCHAR(255) NOT NULL UNIQUE
);

针对每个事件类型描述了更改此表的更改事件的值部分。

3.创建事件

以下示例显示了连接器为在客户表中创建数据的操作生成的更改事件的值部分：

{
  "schema": {    (1)
    "type": "struct",
    "fields": [
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
          },
          {
            "type": "string",
            "optional": false,
            "field": "first_name"
          },
          {
            "type": "string",
            "optional": false,
            "field": "last_name"
          },
          {
            "type": "string",
            "optional": false,
            "field": "email"
          }
        ],
        "optional": true,
        "name": "server1.dbo.customers.Value",  (2)
        "field": "before"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
          },
          {
            "type": "string",
            "optional": false,
            "field": "first_name"
          },
          {
            "type": "string",
            "optional": false,
            "field": "last_name"
          },
          {
            "type": "string",
            "optional": false,
            "field": "email"
          }
        ],
        "optional": true,
        "name": "server1.dbo.customers.Value",
        "field": "after"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "string",
            "optional": false,
            "field": "version"
          },
          {
            "type": "string",
            "optional": false,
            "field": "connector"
          },
          {
            "type": "string",
            "optional": false,
            "field": "name"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "ts_ms"
          },
          {
            "type": "boolean",
            "optional": true,
            "default": false,
            "field": "snapshot"
          },
          {
            "type": "string",
            "optional": false,
            "field": "db"
          },
          {
            "type": "string",
            "optional": false,
            "field": "schema"
          },
          {
            "type": "string",
            "optional": false,
            "field": "table"
          },
          {
            "type": "string",
            "optional": true,
            "field": "change_lsn"
          },
          {
            "type": "string",
            "optional": true,
            "field": "commit_lsn"
          },
          {
            "type": "int64",
            "optional": true,
            "field": "event_serial_no"
          }
        ],
        "optional": false,
        "name": "io.debezium.connector.sqlserver.Source",   (3)
        "field": "source"
      },
      {
        "type": "string",
        "optional": false,
        "field": "op"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_ms"
      }
    ],
    "optional": false,
    "name": "server1.dbo.customers.Envelope"     (4)
  },
  "payload": { (5)
    "before": null, (6)
    "after": { (7)
      "id": 1005,
      "first_name": "john",
      "last_name": "doe",
      "email": "john.doe@example.org"
    },
    "source": {   (8)
      "version": "1.9.6.Final",
      "connector": "sqlserver",
      "name": "server1",
      "ts_ms": 1559729468470,
      "snapshot": false,
      "db": "testDB",
      "schema": "dbo",
      "table": "customers",
      "change_lsn": "00000027:00000758:0003",
      "commit_lsn": "00000027:00000758:0005",
      "event_serial_no": "1"
    },
    "op": "c",     (9)
    "ts_ms": 1559729471739                  (10)
  }
}

表 6. 创建事件值字段的描述

4.更新事件

示例客户表中更新的更改事件的值与该表的创建事件具有相同的模式。同样，事件值的有效负载具有相同的结构。但是，事件值有效负载在更新事件中包含不同的值。以下是连接器为客户表中的更新生成的事件中的更改事件值示例：

{
  "schema": { ... },
  "payload": {
    "before": { (1)
      "id": 1005,
      "first_name": "john",
      "last_name": "doe",
      "email": "john.doe@example.org"
    },
    "after": { (2)
      "id": 1005,
      "first_name": "john",
      "last_name": "doe",
      "email": "noreply@example.org"
    },
    "source": { (3)
      "version": "1.9.6.Final",
      "connector": "sqlserver",
      "name": "server1",
      "ts_ms": 1559729995937,
      "snapshot": false,
      "db": "testDB",
      "schema": "dbo",
      "table": "customers",
      "change_lsn": "00000027:00000ac0:0002",
      "commit_lsn": "00000027:00000ac0:0007",
      "event_serial_no": "2"
    },
    "op": "u", (4)
    "ts_ms": 1559729998706  (5)
  }
}

表 7. 更新事件值字段的描述

注意：


更新行的主键/唯一键的列会更改行键的值。当一个键改变时，Debezium 会输出三个事件：一个删除事件和一个带有该行旧键的墓碑事件，然后是一个带有该行新键的创建事件。

5.删除事件

删除更改事件中的值与同一表的创建和更新事件具有相同的架构部分。示例客户表的删除事件中的有效负载部分如下所示：

{
  "schema": { ... },
  },
  "payload": {
    "before": { <>
      "id": 1005,
      "first_name": "john",
      "last_name": "doe",
      "email": "noreply@example.org"
    },
    "after": null,      (2)
    "source": {          (3)
      "version": "1.9.6.Final",
      "connector": "sqlserver",
      "name": "server1",
      "ts_ms": 1559730445243,
      "snapshot": false,
      "db": "testDB",
      "schema": "dbo",
      "table": "customers",
      "change_lsn": "00000027:00000db0:0005",
      "commit_lsn": "00000027:00000db0:0007",
      "event_serial_no": "1"
    },
    "op": "d",         (4)
    "ts_ms": 1559730450205     (5)
  }
}

表 8. 删除事件值字段的描述

SQL Server 连接器事件旨在与 Kafka 日志压缩一起使用。只要至少保留每个键的最新消息，日志压缩就可以删除一些较旧的消息。这让 Kafka 可以回收存储空间，同时确保主题包含完整的数据集并可用于重新加载基于键的状态。

6.墓碑事件

当删除一行时，删除事件值仍然适用于日志压缩，因为 Kafka 可以删除所有具有相同键的早期消息。但是，要让 Kafka 删除具有相同键的所有消息，消息值必须为空。为了实现这一点，在 Debezium 的 SQL Server 连接器发出删除事件后，连接器会发出一个特殊的墓碑事件，该事件具有相同的键但为空值。

四、事物元数据

Debezium 可以生成表示事务边界和丰富数据更改事件消息的事件。

Debezium 接收事务元数据的时间限制，Debezium 仅注册和接收部署连接器后发生的事务的元数据。部署连接器之前发生的事务的元数据不可用。

数据库事务由包含在 BEGIN 和 END 关键字之间的语句块表示。 Debezium 为每个事务中的 BEGIN 和 END 分隔符生成事务边界事件。事务边界事件包含以下字段：

status

• BEGIN or END

id：

• 唯一交易标识符的字符串表示。

event_count（用于 END 事件）

• 事务发出的事件总数。

data_collections（用于 END 事件）

• data_collection 和 event_count 对的数组，提供源自给定数据集合的更改发出的事件数。

警告：

Debezium 无法可靠地识别交易何时结束。因此，仅在另一个事务的第一个事件到达后才会发出事务结束标记。在低流量系统的情况下，这可能会导致延迟交付 END 标记。

以下示例显示了典型的事务边界消息：

{
  "status": "BEGIN",
  "id": "00000025:00000d08:0025",
  "event_count": null,
  "data_collections": null
}

{
  "status": "END",
  "id": "00000025:00000d08:0025",
  "event_count": 2,
  "data_collections": [
    {
      "data_collection": "testDB.dbo.tablea",
      "event_count": 1
    },
    {
      "data_collection": "testDB.dbo.tableb",
      "event_count": 1
    }
  ]
}

除非通过 transaction.topic 选项覆盖，否则事务事件将写入名为 database.server.name.transaction 的主题。

Change data event enrichment

启用事务元数据后，数据消息 Envelope 会增加一个新事务字段。此字段以字段组合的形式提供有关每个事件的信息：

ID：

• 唯一交易标识符的字符串表示

total_order：

• 该事件在交易产生的所有事件中的绝对位置

data_collection_order

• 该事件在事务发出的所有事件中的每个数据收集位置

以下示例显示了典型消息的外观：

{
  "before": null,
  "after": {
    "pk": "2",
    "aa": "1"
  },
  "source": {
...
  },
  "op": "c",
  "ts_ms": "1580390884335",
  "transaction": {
    "id": "00000025:00000d08:0025",
    "total_order": "1",
    "data_collection_order": "1"
  }
}

五、数据类型映射

Debezium SQL Server 连接器通过生成与行所在的表类似的事件来表示对表行数据的更改。每个事件都包含表示行的列值的字段。事件表示操作的列值的方式取决于列的 SQL 数据类型。在这种情况下，连接器将每个 SQL Server 数据类型的字段映射到文字类型和语义类型。

连接器可以将 SQL Server 数据类型映射到文字和语义类型。

文字类型：

• 描述如何使用 Kafka Connect 模式类型（即 INT8、INT16、INT32、INT64、FLOAT32、FLOAT64、BOOLEAN、STRING、BYTES、ARRAY、MAP 和 STRUCT）字面表示值。

语义类型

• 描述 Kafka Connect 架构如何使用字段的 Kafka Connect 架构名称来捕获字段的含义。

如果默认数据类型转换不能满足您的需求，您可以为连接器创建自定义转换器。

1.基本类型

下表显示了连接器如何映射基本 SQL Server 数据类型。

表 9. SQL Server 连接器使用的数据类型映射

以下部分描述了其他数据类型映射。

如果存在，则将列的默认值传播到相应字段的 Kafka Connect 架构。更改消息将包含该字段的默认值（除非已给出明确的列值），因此很少需要从模式中获取默认值。尽管将 Avro 作为序列化格式与 Confluent 模式注册表一起使用时，传递默认值有助于满足兼容性规则。

2.时间值

除了 SQL Server 的 DATETIMEOFFSET 数据类型（包含时区信息）之外，其他时间类型取决于 time.precision.mode 配置属性的值。当 time.precision.mode 配置属性设置为adaptive（默认）时，连接器将根据列的数据类型定义确定时间类型的文字类型和语义类型，以便事件准确地表示数据库中的值：

当 time.precision.mode 配置属性设置为 connect 时，连接器将使用预定义的 Kafka Connect 逻辑类型。当消费者只知道内置的 Kafka Connect 逻辑类型并且无法处理可变精度的时间值时，这可能很有用。另一方面，由于 SQL Server 支持十分之一微秒精度，因此当数据库列的小数秒精度值大于 3 时，具有连接时间精度模式的连接器生成的事件将导致精度损失：

3.时间戳值

DATETIME、SMALLDATETIME 和 DATETIME2 类型表示没有时区信息的时间戳。此类列将转换为基于 UTC 的等效 Kafka Connect 值。因此，例如 DATETIME2 值“2018-06-20 15:13:16.945104”由值为“1529507596945104”的 io.debezium.time.MicroTimestamp 表示。

请注意，运行 Kafka Connect 和 Debezium 的 JVM 的时区不会影响此转换。

4.十进制值

Debezium 连接器根据 decimal.handling.mode 连接器配置属性的设置处理小数。

decimal.handling.mode=precise

表 10. 当 decimal.handling.mode=precise 时的映射

decimal.handling.mode=double

表 11. 当 decimal.handling.mode=double 时的映射

decimal.handling.mode=string

表 12. 当 decimal.handling.mode=string 时的映射

六、设置SQL Server

为了让 Debezium 从 SQL Server 表中捕获更改事件，具有必要权限的 SQL Server 管理员必须首先运行查询以在数据库上启用 CDC。然后管理员必须为您希望 Debezium 捕获的每个表启用 CDC。

应用 CDC 后，它会捕获提交到启用了 CDD 的表的所有 INSERT、UPDATE 和 DELETE 操作。然后 Debezium 连接器可以捕获这些事件并将它们发送到 Kafka 主题。

1.在 SQL Server 数据库上启用 CDC

在为表启用 CDC 之前，您必须为 SQL Server 数据库启用它。 SQL Server 管理员通过运行系统存储过程来启用 CDC。系统存储过程可以使用 SQL Server Management Studio 或 Transact-SQL 运行。

先决条件

• 您是 SQL Server 的 sysadmin 固定服务器角色的成员。
• 您是数据库的 db_owner。
• SQL Server 代理正在运行。

注意：

• SQL Server CDC 功能仅处理用户创建的表中发生的更改。您不能在 SQL Server 主数据库上启用 CDC。

程序

• 1.从 SQL Server Management Studio 的“查看”菜单中，单击“模板资源管理器”。
• 2.在模板浏览器中，展开 SQL Server 模板。
• 3.展开更改数据捕获 > 配置，然后单击为 CDC 启用数据库。
• 4.在模板中，将 USE 语句中的数据库名称替换为要为 CDC 启用的数据库名称。
• 5.运行存储过程 sys.sp_cdc_enable_db 为 CDC 启用数据库。

为 CDC 启用数据库后，将创建名为 cdc 的模式，以及 CDC 用户、元数据表和其他系统对象。

以下示例显示如何为数据库 MyDB 启用 CDC：

示例：为 CDC 模板启用 SQL Server 数据库

USE MyDB
GO
EXEC sys.sp_cdc_enable_db
GO

2.在 SQL Server 表上启用 CDC

SQL Server 管理员必须在您希望 Debezium 捕获的源表上启用更改数据捕获。数据库必须已为 CDC 启用。要在表上启用 CDC，SQL Server 管理员为该表运行存储过程 sys.sp_cdc_enable_table。存储过程可以使用 SQL Server Management Studio 或 Transact-SQL 运行。必须为要捕获的每个表启用 SQL Server CDC。

先决条件：

• CDC 在 SQL Server 数据库上启用。
• SQL Server 代理正在运行。
• 您是数据库的 db_owner 固定数据库角色的成员。

程序

• 从 SQL Server Management Studio 的“查看”菜单中，单击“模板资源管理器”。
• 在模板浏览器中，展开 SQL Server 模板。
• 展开更改数据捕获 > 配置，然后单击启用表指定文件组选项。
• 在模板中，将 USE 语句中的表名替换为您要捕获的表名。
• 运行存储过程 sys.sp_cdc_enable_table。

以下示例显示如何为表 MyTable 启用 CDC：

示例：为 SQL Server 表启用 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'MyDB_CT',
@supports_net_changes = 0
GO

• 指定要捕获的表的名称。
• 指定角色 MyRole，您可以向该角色添加要授予对源表的捕获列的 SELECT 权限的用户。具有 sysadmin 或 db_owner 角色的用户还可以访问指定的更改表。将 @role_name 的值设置为 NULL，以仅允许 sysadmin 或 db_owner 中的成员对捕获的信息具有完全访问权限。
• 指定 SQL Server 为捕获的表放置更改表的文件组。命名的文件组必须已经存在。最好不要将更改表放在用于源表的同一文件组中。

3.验证用户是否有权访问 CDC 表

SQL Server 管理员可以运行系统存储过程来查询数据库或表以检索其 CDC 配置信息。存储过程可以使用 SQL Server Management Studio 或 Transact-SQL 运行。

先决条件

• 您对捕获实例的所有捕获列具有 SELECT 权限。 db_owner 数据库角色的成员可以查看所有已定义捕获实例的信息。
• 您拥有为查询包括的表信息定义的任何门控角色的成员资格。

程序

• 从 SQL Server Management Studio 的“查看”菜单中，单击“对象资源管理器”。
• 在对象资源管理器中，展开数据库，然后展开您的数据库对象，例如 MyDB。
• 展开可编程性 > 存储过程 > 系统存储过程。
• 运行 sys.sp_cdc_help_change_data_capture 存储过程来查询表。

查询不应返回空结果。

以下示例在数据库 MyDB 上运行存储过程 sys.sp_cdc_help_change_data_capture：

示例：查询表以获取 CDC 配置信息

USE MyDB;
GO
EXEC sys.sp_cdc_help_change_data_capture
GO

该查询返回数据库中每个表的配置信息，这些表为 CDC 启用并且包含调用者有权访问的更改数据。如果结果为空，请验证用户是否具有访问捕获实例和 CDC 表的权限。

七、SQL Server 始终开启

SQL Server 连接器可以从 Always On 只读副本捕获更改。

先决条件

• 在主节点上配置并启用变更数据捕获。 SQL Server 不直接在副本上支持 CDC。

• 配置选项 database.applicationIntent 设置为 ReadOnly。这是 SQL Server 所要求的。当 Debezium 检测到此配置选项时，它会通过以下操作进行响应：

• 
  将snapshot.isolation.mode 设置为snapshot，这是只读副本唯一支持的一种事务隔离模式。

• 
  在每次执行流式查询循环时提交（只读）事务，这是获取 CDC 数据最新视图所必需的。

八、SQL Server 捕获作业代理配置对服务器负载和延迟的影响

当数据库管理员为源表启用变更数据捕获时，捕获作业代理开始运行。代理从事务日志中读取新的更改事件记录，并将事件记录复制到更改数据表中。在源表中提交更改与更改出现在相应更改表中的时间之间，始终存在一个小的延迟间隔。此延迟时间间隔表示源表中发生更改与 Debezium 可用于流式传输到 Apache Kafka 之间的差距。

理想情况下，对于必须快速响应数据更改的应用程序，您希望在源表和更改表之间保持密切同步。您可能会想像，运行捕获代理以尽可能快地连续处理变更事件可能会提高吞吐量并减少延迟 — 在事件发生后尽快用新的事件记录填充变更表，几乎是实时的。但是，情况不一定如此。追求更直接的同步需要付出性能代价。每次捕获作业代理向数据库查询新的事件记录时，都会增加数据库主机上的 CPU 负载。服务器上的额外负载可能会对整体数据库性能产生负面影响，并可能降低事务效率，尤其是在数据库使用高峰期。

监视数据库指标很重要，这样您就可以知道数据库是否达到服务器不再支持捕获代理的活动级别的程度。如果您发现性能问题，您可以修改 SQL Server 捕获代理设置，以帮助平衡数据库主机上的总体 CPU 负载和可容忍的延迟程度。

九、SQL Server 捕获作业代理配置参数

在 SQL Server 上，控制捕获作业代理行为的参数在 SQL Server 表 msdb.dbo.cdc_jobs 中定义。如果您在运行捕获作业代理时遇到性能问题，请通过运行 sys.sp_cdc_change_job 存储过程并提供新值来调整捕获作业设置以减少 CPU 负载。

注意：

• 有关如何配置 SQL Server 捕获作业代理参数的具体指导超出了本文档的范围。

以下参数对于修改用于 Debezium SQL Server 连接器的捕获代理行为最重要：

轮询间隔：

• 指定捕获代理在日志扫描周期之间等待的秒数。
• 较高的值会减少数据库主机上的负载并增加延迟。
• 值 0 指定扫描之间不等待。
• 默认值为 5。

最大传输：

• 指定在每个日志扫描周期内处理的最大事务数。在捕获作业处理指定数量的事务后，它会暂停 pollinginterval 指定的时间长度，然后再开始下一次扫描。
• 较低的值会减少数据库主机上的负载并增加延迟。
• 默认值为 500。

最大扫描数：

• 指定捕获作业在捕获数据库事务日志的全部内容时可以尝试的扫描周期数限制。如果连续参数设置为 1，则作业会在 pollinginterval 指定的时间长度内暂停，然后再继续扫描。
• 较低的值会减少数据库主机上的负载并增加延迟。
• 默认值为 10。

有关捕获代理参数的详细信息，请参阅 SQL Server 文档。

十、部署

要部署 Debezium SQL Server 连接器，请安装 Debezium SQL Server 连接器存档，配置连接器，然后通过将其配置添加到 Kafka Connect 来启动连接器。

先决条件

• 已安装 Apache ZooKeeper、Apache Kafka 和 Kafka Connect。
• SQL Server 已安装，已针对 CDC 进行配置，并且可以与 Debezium 连接器一起使用。

程序

• 下载 Debezium SQL Server 连接器插件存档
• 将文件提取到您的 Kafka Connect 环境中。
• 将包含 JAR 文件的目录添加到 Kafka Connect 的 plugin.path。
• 配置连接器并将配置添加到您的 Kafka Connect 集群。
• 重新启动 Kafka Connect 进程以获取新的 JAR 文件。

详细的部署过程记录在下面这篇博文中：

• Debezium系列之：安装部署debezium详细步骤，并把debezium服务托管到systemctl
  
  https://blog.csdn.net/zhengzaifeidelushang/article/details/124408909
  

十一、SQL Server 连接器配置示例

以下是连接器实例的配置示例，该实例在 192.168.99.100 的端口 1433 上从 SQL Server 服务器捕获数据，我们在逻辑上将其命名为 fullfillment。通常，您通过设置可用于连接器的配置属性，在 JSON 文件中配置 Debezium SQL Server 连接器。

您可以选择为数据库中的模式和表的子集生成事件。或者，您可以忽略、屏蔽或截断包含敏感数据、大于指定大小或您不需要的列。

{
    "name": "inventory-connector",     (1)
    "config": {
        "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",     (2)
        "database.hostname": "192.168.99.100",       (3)
        "database.port": "1433",                      (4)
        "database.user": "sa",                        (5)
        "database.password": "Password!",              (6)
        "database.dbname": "testDB",                   (7)
        "database.server.name": "fullfillment",       (8)
        "table.include.list": "dbo.customers",        (9)
        "database.history.kafka.bootstrap.servers": "kafka:9092", (10)
        "database.history.kafka.topic": "dbhistory.fullfillment"    (11)
    }
}

• 1.当我们向 Kafka Connect 服务注册连接器时的名称。
• 2.此 SQL Server 连接器类的名称。
• 3.SQL Server 实例的地址。
• 4.SQL Server 实例的端口号。
• 5.SQL Server 用户的名称
• 6.SQL Server 用户的密码
• 7.要从中捕获更改的数据库的名称。
• 8.SQL Server 实例/集群的逻辑名称，形成一个命名空间，用于连接器写入的所有 Kafka 主题名称、Kafka Connect 架构名称以及 Avro 转换器时对应 Avro 架构的命名空间用来。
• 9.Debezium 应捕获其更改的所有表的列表。
• 10.此连接器将用于将 DDL 语句写入和恢复到数据库历史主题的 Kafka 代理列表。
• 11.连接器将写入和恢复 DDL 语句的数据库历史主题的名称。本主题仅供内部使用，消费者不得使用。

有关可以为 Debezium SQL Server 连接器设置的配置属性的完整列表，请参阅 SQL Server 连接器属性。

您可以使用 POST 命令将此配置发送到正在运行的 Kafka Connect 服务。该服务记录配置并启动执行以下任务的一个连接器任务：

• 连接到 SQL Server 数据库。
• 读取事务日志。
• 将更改事件记录到 Kafka 主题。

添加连接器配置

要开始运行 Debezium SQL Server 连接器，请创建连接器配置，并将配置添加到您的 Kafka Connect 集群。

先决条件

• CDC 在 SQL Server 上启用。
• Debezium SQL Server 连接器已安装。

程序

• 为 SQL Server 连接器创建配置。
• 使用 Kafka Connect REST API 将该连接器配置添加到您的 Kafka Connect 集群。

结果

当连接器启动时，它会为连接器配置的 SQL Server 数据库执行一致的快照。然后，连接器开始为行级操作生成数据更改事件，并将更改事件记录流式传输到 Kafka 主题。

十二、连接器属性

Debezium SQL Server 连接器具有许多配置属性，您可以使用这些属性为您的应用程序实现正确的连接器行为。许多属性都有默认值。

有关属性的信息组织如下：

• 必需的连接器配置属性
• 高级连接器配置属性
• 控制 Debezium 如何处理从数据库历史主题中读取的事件的数据库历史连接器配置属性。
• 传递数据库历史记录属性
• 控制数据库驱动程序行为的传递数据库驱动程序属性。

1.必需的 Debezium SQL Server 连接器配置属性

除非默认值可用，否则需要以下配置属性。

2.高级 SQL Server 连接器配置属性

以下高级配置属性具有良好的默认值，适用于大多数情况，因此很少需要在连接器的配置中指定。

3.Debezium SQL Server 连接器数据库历史配置属性

Debezium 提供了一组 database.history.* 属性来控制连接器如何与模式历史主题交互。

下表描述了用于配置 Debezium 连接器的 database.history 属性。

4.用于配置生产者和消费者客户端的直通数据库历史记录属性

Debezium 依赖 Kafka 生产者将模式更改写入数据库历史主题。同样，当连接器启动时，它依赖于 Kafka 消费者从数据库历史主题中读取。您可以通过将值分配给以 database.history.producer.* 和 database.history.consumer.* 前缀开头的一组传递配置属性来定义 Kafka 生产者和消费者客户端的配置。传递的生产者和消费者数据库历史属性控制一系列行为，例如这些客户端如何保护与 Kafka 代理的连接，如以下示例所示：

database.history.producer.security.protocol=SSL
database.history.producer.ssl.keystore.location=/var/private/ssl/kafka.server.keystore.jks
database.history.producer.ssl.keystore.password=test1234
database.history.producer.ssl.truststore.location=/var/private/ssl/kafka.server.truststore.jks
database.history.producer.ssl.truststore.password=test1234
database.history.producer.ssl.key.password=test1234

database.history.consumer.security.protocol=SSL
database.history.consumer.ssl.keystore.location=/var/private/ssl/kafka.server.keystore.jks
database.history.consumer.ssl.keystore.password=test1234
database.history.consumer.ssl.truststore.location=/var/private/ssl/kafka.server.truststore.jks
database.history.consumer.ssl.truststore.password=test1234
database.history.consumer.ssl.key.password=test1234

Debezium 在将属性传递给 Kafka 客户端之前从属性名称中去除前缀​​。

有关 Kafka 生产者配置属性和 Kafka 消费者配置属性的更多详细信息，请参阅 Kafka 文档。

5.Debezium SQL Server 连接器直通数据库驱动程序配置属性

Debezium 连接器提供数据库驱动程序的直通配置。直通数据库属性以前缀 database.* 开头。例如，连接器将诸如 database.foobar=false 之类的属性传递给 JDBC URL。

与数据库历史客户端的传递属性一样，Debezium 在将它们传递给数据库驱动程序之前从属性中去除前缀​​。

十三、数据库模式演变

当为 SQL Server 表启用更改数据捕获时，当表中发生更改时，事件记录将持久保存到服务器上的捕获表中。如果您在源表更改的结构中引入更改，例如，通过添加新列，则该更改不会动态反映在更改表中。只要捕获表继续使用过时的模式，Debezium 连接器就无法正确地为表发出数据更改事件。您必须进行干预以刷新捕获表，以使连接器能够恢复处理更改事件。

由于 CDC 在 SQL Server 中的实现方式，您不能使用 Debezium 更新捕获表。要刷新捕获表，必须是具有提升权限的 SQL Server 数据库操作员。作为 Debezium 用户，您必须与 SQL Server 数据库操作员协调任务，以完成架构刷新并将流恢复到 Kafka 主题。

您可以使用以下方法之一在架构更改后更新捕获表：

• 离线模式更新要求您在更新捕获表之前停止 Debezium 连接器。
• 在线模式更新可以在 Debezium 连接器运行时更​​新捕获表。

使用每种类型的程序都有优点和缺点。

警告：

无论您使用在线更新方法还是离线更新方法，都必须先完成整个架构更新过程，然后才能对同一个源表应用后续架构更新。最佳实践是在一个批次中执行所有 DDL，这样该过程只能运行一次。

注意：

无论您使用在线更新方法还是离线更新方法，都必须先完成整个架构更新过程，然后才能对同一个源表应用后续架构更新。最佳实践是在一个批次中执行所有 DDL，这样该过程只能运行一次。

将源表中的列从 NULL 更改为 NOT NULL 或反之亦然后，SQL Server 连接器无法正确捕获更改的信息，直到您创建新的捕获实例之后。如果您在更改列名称后没有创建新的捕获表，则连接器发出的更改事件记录不会正确指示该列是否是可选的。也就是说，以前定义为可选（或 NULL）的列继续存在，尽管现在被定义为 NOT NULL。同样，已定义为必需的列 (NOT NULL) 保留该名称，尽管它们现在定义为 NULL。

1.离线架构更新

离线模式更新为更新捕获表提供了最安全的方法。但是，离线更新可能不适用于需要高可用性的应用程序。

先决条件

• 已将更新提交到启用了 CDC 的 SQL Server 表的架构。
• 您是具有提升权限的 SQL Server 数据库操作员。

程序

• 挂起更新数据库的应用程序。
• 等待 Debezium 连接器流式传输所有未流式传输的更改事件记录。
• 停止 Debezium 连接器。
• 将所有更改应用于源表架构。
• 使用带有参数@capture_instance 的唯一值的sys.sp_cdc_enable_table 过程为更新源表创建一个新的捕获表。
• 恢复您在步骤 1 中暂停的应用程序。
• 启动 Debezium 连接器。
• 在 Debezium 连接器开始从新的捕获表进行流式传输后，通过运行存储过程 sys.sp_cdc_disable_table 并将参数 @capture_instance 设置为旧的捕获实例名称来删除旧的捕获表。

2.在线模式更新

完成在线模式更新的过程比运行离线模式更新的过程更简单，您可以在不需要任何应用程序和数据处理停机时间的情况下完成它。但是，对于在线模式更新，在您更新源数据库中的模式之后，但在您创建新的捕获实例之前，可能会出现潜在的处理差距。在该间隔期间，更改表的旧实例继续捕获更改事件，并且保存到旧表的更改数据保留了早期模式的结构。因此，例如，如果您向源表中添加了新列，则在新捕获表准备好之前生成的更改事件不包含新列的字段。如果您的应用程序不能容忍这样的过渡期，最好使用离线模式更新过程。

先决条件

• 已将更新提交到启用了 CDC 的 SQL Server 表的架构。
• 您是具有提升权限的 SQL Server 数据库操作员。

程序

• 将所有更改应用于源表架构。
• 通过使用参数@capture_instance 的唯一值运行sys.sp_cdc_enable_table 存储过程，为更新源表创建一个新的捕获表。
• 当 Debezium 从新的捕获表开始流式传输时，您可以通过运行 sys.sp_cdc_disable_table 存储过程并将参数 @capture_instance 设置为旧的捕获实例名称来删除旧的捕获表。

示例：在数据库架构更改后运行在线架构更新

让我们部署基于 SQL Server 的 Debezium 教程来演示在线模式更新。

在以下示例中，将 phone_number 列添加到客户表中。

键入以下命令以启动数据库 shell：

docker-compose -f docker-compose-sqlserver.yaml exec sqlserver bash -c '/opt/mssql-tools/bin/sqlcmd -U sa -P $SA_PASSWORD -d testDB'

通过运行以下查询以添加 phone_number 字段来修改客户源表的架构：

ALTER TABLE customers ADD phone_number VARCHAR(32);

通过运行 sys.sp_cdc_enable_table 存储过程创建新的捕获实例。

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 INTO customers(first_name,last_name,email,phone_number) VALUES ('John','Doe','john.doe@example.com', '+1-555-123456');
GO

Kafka Connect 日志通过类似于以下消息的条目报告配置更新：

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]

最终，phone_number 字段被添加到模式中，并且它的值出现在写入 Kafka 主题的消息中。

...
     {
        "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"
    },

通过运行 sys.sp_cdc_disable_table 存储过程删除旧的捕获实例。

EXEC sys.sp_cdc_disable_table @source_schema = 'dbo', @source_name = 'dbo_customers', @capture_instance = 'dbo_customers';
GO

十四、监控

除了 Zookeeper、Kafka 和 Kafka Connect 提供的对 JMX 指标的内置支持之外，Debezium SQL Server 连接器还提供三种类型的指标。连接器提供以下指标：

用于在执行快照时监控连接器的快照指标。

读取 CDC 表数据时用于监控连接器的流式指标。

用于监控连接器架构历史状态的架构历史指标。

有关如何通过 JMX 公开上述指标的信息，请参阅 Debezium 监控文档。

1.Snapshot metrics快照指标

除非快照操作处于活动状态，或者自上次连接器启动以来发生了快照，否则不会公开快照指标。

• MBean is debezium.sql_server:type=connector-metrics,server=<sqlserver.server.name>,task=<task.id>,context=snapshot.

执行增量快照时，连接器还提供以下附加快照指标：

2.Streaming metrics流指标

MBean is debezium.sql_server:type=connector-metrics,server=<sqlserver.server.name>,task=<task.id>,context=streaming.

Debezium SQL Server 连接器还提供以下额外的流指标：

3.Schema history metrics架构历史指标

MBean is debezium.sql_server:type=connector-metrics,context=schema-history,server=<sqlserver.server.name>.

十五、Debezium应用延伸与扩展

Debezium系列之第100篇文章：阶段性详细总结对Debezium使用方式的优化，详细介绍对Debezium集群和Kafka集群做的一系列优化：

• 
  https://blog.csdn.net/zhengzaifeidelushang/article/details/127044172
  

Debezium系列之：安装部署debezium详细步骤，并把debezium服务托管到systemctl：

• 
  https://blog.csdn.net/zhengzaifeidelushang/article/details/124408909
  

Debezium系列之：多个数据库多张表或单个数据库多张表数据导入同一个kafka topic，新增字段区别数据来自哪个数据库哪张表：

• 
  https://blog.csdn.net/zhengzaifeidelushang/article/details/127054983
  

更多Debezium系列文章，请参考博主Debezium分享专栏：

• 
  https://blog.csdn.net/zhengzaifeidelushang/category_11240845.html?spm=1001.2014.3001.5482