8.2. Debezium PostgreSQL 连接器的工作方式

要使用 Debezium 连接器从 PostgreSQL 数据库流更改，连接器必须使用数据库中的特定权限运行。虽然授予必要的特权的一种方法是为用户提供 超级用户特权，这样做可能会将您的 PostgreSQL 数据暴露给未经授权的访问。最好创建一个专用的 Debezium 复制用户，而不是为 Debezium 用户授予特定的特权。

有关为 Debezium PostgreSQL 用户配置特权的更多信息，请参阅 设置权限。有关 PostgreSQL 逻辑复制安全性的更多信息，请参阅 PostgreSQL 文档。

大多数 PostgreSQL 服务器都配置为不保留 WAL 段中数据库的完整历史记录。这意味着 PostgreSQL 连接器无法通过只读取 WAL 来查看数据库的完整历史记录。因此，连接器首次启动时，它会执行数据库的初始 一致快照。

您可以在以下部分找到有关快照的更多信息：

如果连接器失败，会在步骤 1 开始后重新平衡或停止，但在重启连接器开始新快照前。连接器完成其初始快照后，PostgreSQL 连接器会从在第 2 步中读取的位置继续流。这样可确保连接器不会错过任何更新。如果连接器因为任何原因而再次停止，则在重启时，连接器会继续从之前关闭的位置进行流更改。

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

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

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

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

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

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

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

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

在增量快照中，除了一次捕获数据库的完整状态，就像初始快照一样，Debebe 会在一系列可配置的块中捕获每个表。您可以指定您希望快照捕获的表 以及每个块的大小。块大小决定了快照在数据库的每个获取操作期间收集的行数。增量快照的默认块大小为 1024 行。

当增量快照进行时，Debebe 使用 watermarks 跟踪其进度，维护它捕获的每个表行的记录。与标准初始快照过程相比，捕获数据的阶段方法具有以下优点：

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

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

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

连接器为每个快照块重复这个过程。

SELECT * FROM <tableName> ....

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

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

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

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"}');

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

Key = `test_connector`

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

Key = `test_connector`

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

Key = `test_connector`

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

Key = `test_connector`

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

PostgreSQL 连接器通常将其大部分的时间流更改从 PostgreSQL 服务器连接到其中。这种机制依赖于 PostgreSQL 的复制协议。这个协议可让客户端从服务器接收更改，因为它们在服务器的事务日志中提交，这些位置被称为 Log Sequence Numbers (LSN)。

每当服务器提交事务时，单独的服务器进程会从 逻辑解码插件 调用回调功能。此功能处理事务中的更改，将其转换为特定格式（如果是 Debezium 插件则为Protobuf 或 JSON），并在输出流上写入它们，然后可以被客户端使用。

Debezium PostgreSQL 连接器充当 PostgreSQL 客户端。当连接器收到更改时，它会将事件转换为 Debezium 的 create, update, 或 delete 事件，包含该事件的 LSN 的事件。PostgreSQL 连接器将记录中的这些更改事件转发到 Kafka Connect 框架，该框架在同一进程中运行。Kafka Connect 进程异步写入更改事件记录，其顺序与生成到适当的 Kafka 主题相同。

定期，Kafka Connect 在另一个 Kafka 主题中记录最新的 偏移量。偏移表示 Debezium 包含在每个事件中的特定于源的位置信息。对于 PostgreSQL 连接器，在每次更改事件中记录的 LSN 是偏移量。

当 Kafka Connect 正常关闭时，它会停止连接器，将所有事件记录刷新到 Kafka，并记录从每个连接器接收的最后偏移。当 Kafka Connect 重启时，它会为每个连接器读取最后记录的偏移，并在其最后记录的偏移上启动每个连接器。当连接器重启时，它会将请求发送到 PostgreSQL 服务器，来仅在该位置后开始发送事件。

如需更多信息，请参阅设置 PostgreSQL。

默认情况下，PostgreSQL 连接器将所有 INSERT、UPDATE 和 DELETE 操作的更改事件写入特定于该表的单一 Apache Kafka 主题。连接器使用以下惯例来命名更改事件主题：

topicPrefix.schemaName.tableName

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

例如，假设 fulfillment 是连接器中的逻辑服务器名称，该连接器捕获 PostgreSQL 安装中的更改，该安装具有一个 postgres 数据库和一个 inventory schem，它包含四个表：products, products_on_hand, customers, 和 orders连接器会将记录流传输到这四个 Kafka 主题：

现在假设表不是特定架构的一部分，但在默认的 公共 PostgreSQL 模式中创建。Kafka 主题的名称为：

连接器应用类似的命名约定，以标记其 事务元数据主题。

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

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

对于每个事务 BEGIN 和 END，Debezium 会生成一个包含以下字段的事件：

{
  "status": "BEGIN",
  "id": "571:53195829",
  "ts_ms": 1486500577125,
  "event_count": null,
  "data_collections": null
}

{
  "status": "END",
  "id": "571:53195832",
  "ts_ms": 1486500577691,
  "event_count": 2,
  "data_collections": [
    {
      "data_collection": "s1.a",
      "event_count": 1
    },
    {
      "data_collection": "s2.a",
      "event_count": 1
    }
  ]
}

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

以下是消息的示例：

{
  "before": null,
  "after": {
    "pk": "2",
    "aa": "1"
  },
  "source": {
   ...
  },
  "op": "c",
  "ts_ms": "1580390884335",
  "transaction": {
    "id": "571:53195832",
    "total_order": "1",
    "data_collection_order": "1"
  }
}