8.2. Debezium SQL Server 连接器如何工作

默认情况下，连接器仅在首次启动后运行初始快照操作。在正常情况下，遵循这个初始快照，连接器不会重复快照过程。连接器捕获的任何将来更改事件数据都仅通过流传输过程。

然而，在某些情况下，连接器在初始快照期间获取的数据可能会变得过时、丢失或不完整。为了提供总结表数据的机制，Debezium 包含一个执行临时快照的选项。数据库中的以下更改可能是执行临时快照：

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

当您启动现有表的临时快照时，连接器会将内容附加到表已存在的主题中。如果删除了之前存在的主题，如果启用自动主题创建，Debezium 可以自动创建主题。https://access.redhat.com/documentation/zh-cn/red_hat_integration/2023.q2/html-single/debezium_user_guide/index#customization-of-kafka-connect-automatic-topic-creation

临时快照信号指定要包含在快照中的表。快照可以捕获数据库的整个内容，或者只捕获数据库中表的子集。另外，快照也可以捕获数据库中表内容的子集。

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

目前，exec-snapshot 操作类型仅触发 增量快照。如需更多信息，请参阅 增加快照。

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

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

当增量快照进行时，Debezium 使用水位线来跟踪其进度，维护它捕获的每一个表行的记录。这个阶段捕获数据的方法比标准初始快照过程提供以下优点：

当快照继续进行时，其他进程可能会继续访问数据库，从而可能会修改表记录。要反映此类更改，INSERT、UPDATE 或 DELETE 操作会照常提交到事务日志中。同样，持续 Debezium 流过程还会继续检测这些更改事件，并将对应的更改事件记录发送到 Kafka。

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

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

连接器重复每个快照块的进程。

目前，启动增量快照的唯一方法是向源数据库上的 信号发送临时快照 信号。

您可以以 SQL INSERT 查询形式向信号表提交信号。

在 Debezium 检测到信号表中的更改后，它会读取信号，并运行请求的快照操作。

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

要指定快照中包含的表，请提供一个 data-collections 数组，该数组列出了用于匹配表的表或一组正则表达式，例如：

{"data-collections": ["public.MyFirstTable", "public.MySecondTable"]}

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

典型的快照的 SQL 查询采用以下格式：

SELECT * FROM <tableName> ....

通过添加 additional-condition 参数，您可以将 WHERE 条件附加到 SQL 查询中，如下例所示：

SELECT * FROM <tableName> WHERE <additional-condition> ....

以下示例显示了向信号表发送带有额外条件的临时增量快照请求的 SQL 查询：

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

例如，假设您有一个包含以下列的产品表：

如果您希望 product 表的增量快照只包含 color=blue 的数据项，您可以使用以下 SQL 语句来触发快照：

INSERT INTO myschema.debezium_signal (id, type, data) VALUES('ad-hoc-1', 'execute-snapshot', '{"data-collections": ["schema1.products"],"type":"incremental", "additional-condition":"color=blue"}');

additional-condition 参数还允许您传递基于多个列的条件。例如，使用上例中的 product 表，您可以提交查询来触发增量快照，该快照仅包含那些项目的 data =blue 和 quantity>10 ：

INSERT INTO myschema.debezium_signal (id, type, data) VALUES('ad-hoc-1', 'execute-snapshot', '{"data-collections": ["schema1.products"],"type":"incremental", "additional-condition":"color=blue AND quantity>10"}');

以下示例显示了连接器捕获的增量快照事件的 JSON。

{
    "before":null,
    "after": {
        "pk":"1",
        "value":"New data"
    },
    "source": {
        ...
        "snapshot":"incremental" 1
    },
    "op":"r", 2
    "ts_ms":"1620393591654",
    "transaction":null
}

您还可以通过向源数据库上的表发送信号来停止增量快照。您可以通过发送 SQL INSERT 查询，向表提交停止快照信号。

在 Debezium 检测到信号表中的更改后，它会读取信号，并在进行时停止增量快照操作。

您提交的查询指定了 增量 的快照操作，以及可选的、要删除的当前运行快照的表。

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

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

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
);

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

更改事件中的值比键稍微复杂。与键一样，该值具有 schema 部分和 payload 部分。schema 部分包含描述 payload 部分的 Envelope 结构的 schema，包括其嵌套字段。更改创建、更新或删除数据的事件，并且具有 envelope 结构的值有效负载。

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

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
);

为每个事件类型描述了更改事件的值部分。

{
  "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.testDB.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.testDB.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.testDB.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": "2.1.4.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
  }
}

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

{
  "schema": { ... },
  },
  "payload": {
    "before": { <>
      "id": 1005,
      "first_name": "john",
      "last_name": "doe",
      "email": "noreply@example.org"
    },
    "after": null, 1
    "source": { 2
      "version": "2.1.4.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", 3
    "ts_ms": 1559730450205 4
  }
}

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

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

以下示例显示了典型的信息是什么：

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