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

连接器运行的初始快照捕获两种类型的信息：

运行初始快照后，您可能会注意到快照捕获没有指定用于捕获的表的模式信息。默认情况下，初始快照旨在捕获数据库中存在的每个表的模式信息，而不仅仅是从指定为捕获的表的表。连接器要求表的模式存在于架构历史记录主题中，然后才能捕获表。通过启用初始快照来捕获不是原始捕获集一部分的表的 schema 数据，Debebe 准备好连接器，以便稍后需要捕获这些表中的事件数据。如果初始快照没有捕获表的 schema，您必须将模式添加到历史记录主题，然后才能从表中捕获数据。

在某些情况下，您可能想要限制初始快照中的模式捕获。当您要减少完成快照所需的时间时，这非常有用。或者，当 Debezium 通过可访问多个逻辑数据库的用户帐户连接到数据库实例时，但您希望连接器只从特定逻辑数据库中的表捕获更改。

在某些情况下，您可能希望连接器从其模式未被初始快照捕获的表中捕获数据。根据连接器配置，初始快照只能捕获数据库中特定表的表模式。如果历史记录主题中没有表模式，连接器将无法捕获表，并报告缺少的 schema 错误。

您可能仍然能够从表中捕获数据，但您必须执行额外的步骤来添加表模式。

如果架构更改应用到表，则在架构更改前提交的记录与更改后提交的不同结构不同。当 Debezium 从表中捕获数据时，它会读取 schema 历史记录，以确保它为每个事件应用正确的模式。如果 schema 历史记录主题中没有 schema，则连接器无法捕获表，并出现错误结果。

如果要从初始快照捕获的表中捕获数据，并且修改了表的 schema，则必须将模式添加到历史记录主题中（如果它还没有可用）。您可以通过运行新的模式快照或运行表的初始快照来添加模式。

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

作为 SQL INSERT 查询，您将向信号提交信号。

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

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

要指定快照中包含的表，请提供列出表或用于匹配表的正则表达式数组的 数据集合，例如：

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

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

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

SELECT * FROM <tableName> ....

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

SELECT * FROM <data-collection> WHERE <filter> ....

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

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

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

如果您需要 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-conditions":[{"data-collection": "schema1.products", "filter": "color=blue"}]}');

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

INSERT INTO myschema.debezium_signal (id, type, data) VALUES('ad-hoc-1', 'execute-snapshot', '{"data-collections": ["schema1.products"],"type":"incremental", "additional-conditions":[{"data-collection": "schema1.products", "filter": "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
}

您可以向 配置的 Kafka 主题 发送消息，以请求连接器来运行临时增量快照。

Kafka 消息的键必须与 topic.prefix 连接器配置选项的值匹配。

message 的值是带有 type 和 data 字段的 JSON 对象。

信号类型是 execute-snapshot，data 字段必须具有以下字段：

execute-snapshot Kafka 消息示例：

Key = `test_connector`

Value = `{"type":"execute-snapshot","data": {"data-collections": ["schema1.table1", "schema1.table2"], "type": "INCREMENTAL"}}`

通常，当 Debezium 运行快照时，它会运行 SQL 查询，例如：

SELECT * FROM <tableName> …​.

当快照请求包含 additional-conditions 属性时，属性的 data-collection 和 filter 参数会附加到 SQL 查询中，例如：

SELECT * FROM &lt ;data-collection> WHERE & lt;filter> …​.

例如，如果一个带有字段 id （主键）、颜色 和品牌 的产品 表，如果您希望快照只包含 color='blue' 的内容，当请求快照时，您可以添加 additional-conditions 属性来过滤内容：

Key = `test_connector`

Value = `{"type":"execute-snapshot","data": {"data-collections": ["schema1.products"], "type": "INCREMENTAL", "additional-conditions": [{"data-collection": "schema1.products" ,"filter":"color='blue'"}]}}`

您可以使用 additional-conditions 属性根据多个列传递条件。例如，如果您希望快照只包含 color='blue' 的 products 表中，以及 brand='MyBrand'，则您可以发送以下请求：

Key = `test_connector`

Value = `{"type":"execute-snapshot","data": {"data-collections": ["schema1.products"], "type": "INCREMENTAL", "additional-conditions": [{"data-collection": "schema1.products" ,"filter":"color='blue' AND brand='MyBrand'"}]}}`

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

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

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

您可以将信号消息发送到 配置的 Kafka 信号主题，以停止临时增量快照。

Kafka 消息的键必须与 topic.prefix 连接器配置选项的值匹配。

message 的值是带有 type 和 data 字段的 JSON 对象。

信号类型是 stop-snapshot，data 字段必须具有以下字段：

以下示例显示了典型的 stop-snapshot Kafka 信息：

Key = `test_connector`

Value = `{"type":"stop-snapshot","data": {"data-collections": ["schema1.table1", "schema1.table2"], "type": "INCREMENTAL"}}`

为了提供管理快照的灵活性，Debebe 包含了附加临时快照机制，称为 阻塞快照。阻塞快照依赖于 Debezium 机制 向 Debezium 连接器发送信号。

阻塞快照的行为就像 初始快照 一样，但您可以在运行时触发它。

您可能想要运行阻塞快照，而不是在以下情况下使用标准初始快照过程：

例如：

  {"type": "blocking", "data-collections": ["schema1.table1", "schema1.table2"], "additional-conditions": [{"data-collection": "schema1.table1", "filter": "SELECT * FROM [schema1].[table1] WHERE column1 = 0 ORDER BY column2 DESC"}, {"data-collection": "schema1.table2", "filter": "SELECT * FROM [schema1].[table2] WHERE column2 > 0"}]}

更改事件的密钥包含更改表的密钥和更改行的实际键的 schema。当连接器创建事件时，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.5.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.5.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.5.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"
  }
}