Debezium系列之：PostgreSQL数据库的Debezium连接器

• Debezium PostgreSQL 连接器捕获 PostgreSQL 数据库模式中的行级更改。
• 第一次连接到 PostgreSQL 服务器或集群时，连接器会拍摄所有模式的一致快照。在该快照完成后，连接器会持续捕获插入、更新和删除数据库内容以及提交到 PostgreSQL 数据库的行级更改。连接器生成数据更改事件记录并将它们流式传输到 Kafka 主题。对于每个表，默认行为是连接器将所有生成的事件流式传输到该表的单独 Kafka 主题。应用程序和服务使用来自该主题的数据更改事件记录。

一、概述

• PostgreSQL 的逻辑解码功能是在 9.4 版本中引入的。它是一种允许提取提交到事务日志的更改并在输出插件的帮助下以用户友好的方式处理这些更改的机制。输出插件使客户端能够使用更改。

PostgreSQL 连接器包含两个主要部分，它们协同工作以读取和处理数据库更改：

• 一个逻辑解码输出插件。您可能需要安装您选择使用的输出插件。在运行 PostgreSQL 服务器之前，您必须配置一个使用您选择的输出插件的复制槽。插件可以是以下之一：

• decoderbufs 基于 Protobuf 并由 Debezium 社区维护。

• wal2json 基于 JSON 并由 wal2json 社区维护（已弃用，计划在 Debezium 2.0 中删除）。

• pgoutput 是 PostgreSQL 10+ 中的标准逻辑解码输出插件。它由 PostgreSQL 社区维护，并由 PostgreSQL 本身用于逻辑复制。此插件始终存在，因此无需安装其他库。Debezium 连接器将原始复制事件流直接解释为更改事件。

• 读取所选逻辑解码输出插件产生的更改的 Java 代码（实际的 Kafka Connect 连接器）。它使用 PostgreSQL 的流复制协议，通过 PostgreSQL JDBC 驱动程序

相关链接：

• Streaming Replication Protocol Interface
• PostgreSQL JDBC driver

连接器为捕获的每个行级插入、更新和删除操作生成一个更改事件，并为单独的 Kafka 主题中的每个表发送更改事件记录。客户端应用程序读取数据库表对应的Kafka 主题，并且对这些主题接收到的每个行级事件做出反应。

PostgreSQL 通常会在一段时间后清除 write-ahead log (WAL) 段。这意味着连接器没有对数据库所做的所有更改的完整历史记录。因此，当 PostgreSQL 连接器第一次连接到特定的 PostgreSQL 数据库时，它首先对每个数据库模式执行一致的快照。连接器完成快照后，它会从创建快照的确切点继续流式传输更改。这样，连接器从所有数据的一致视图开始，并且不会忽略在拍摄快照时所做的任何更改。

连接器可以容忍故障。当连接器读取更改并产生事件时，它会记录每个事件的 WAL 位置。如果连接器因任何原因（包括通信故障、网络问题或崩溃）停止，则在重新启动时，连接器会继续读取上次停止的 WAL。这包括快照。如果连接器在快照期间停止，则连接器在重新启动时会开始新的快照。

注意事项：

连接器依赖并反映了 PostgreSQL 逻辑解码特性，该特性有以下限制：

• 逻辑解码不支持 DDL 更改。这意味着连接器无法将 DDL 更改事件报告回消费者。
• 只有主服务器支持逻辑解码复制槽。当存在 PostgreSQL 服务器集群时，连接器只能在活动的主服务器上运行。它不能在热备份或热备份副本上运行。如果主服务器出现故障或降级，连接器将停止。主服务器恢复后，您可以重新启动连接器。如果已将其他 PostgreSQL 服务器提升为主服务器，请在重新启动连接器之前调整连接器配置。
• Debezium 目前仅支持使用 UTF-8 字符编码的数据库。使用单字节字符编码，无法正确处理包含扩展 ASCII 代码字符的字符串。

二、连接器的工作原理

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

1.安全

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

2.快照

大多数 PostgreSQL 服务器配置为不在 WAL 段中保留数据库的完整历史记录。这意味着 PostgreSQL 连接器将无法通过仅读取 WAL 来查看数据库的整个历史记录。因此，连接器第一次启动时，它会执行数据库的初始一致快照。执行快照的默认行为包括以下步骤。您可以通过将 snapshot.mode 连接器配置属性设置为initial以外的值来更改此行为。

1. 使用 SERIALIZABLE、READ ONLY、DEFERRABLE 隔离级别启动事务，以确保此事务中的后续读取针对数据的单个一致版本。由于其他客户端的后续 INSERT、UPDATE 和 DELETE 操作而对数据进行的任何更改对此事务不可见。
2. 读取服务器事务日志中的当前位置。
3. 扫描数据库表和模式，为每一行生成一个 READ 事件并将该事件写入适当的特定于表的 Kafka 主题。
4. 提交事务。
5. 在连接器偏移中记录快照的成功完成。

如果连接器出现故障、重新平衡或在步骤 1 开始之后但在步骤 5 完成之前停止，则在重新启动时连接器将开始一个新的快照。连接器完成其初始快照后，PostgreSQL 连接器继续从它在步骤 2 中读取的位置进行流式传输。这可确保连接器不会错过任何更新。如果连接器由于任何原因再次停止，则在重新启动时，连接器会继续从之前停止的位置流式传输更改。

表 1. snapshot.mode 连接器配置属性的选项

3.Ad hoc snapshots

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

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

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

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

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

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

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

表 2. ad hoc执行快照信号记录示例

4. 触发hoc snapshot

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

目前，执行快照操作类型仅触发增量快照。

5.增量快照

为了提供管理快照的灵活性，Debezium 包含一个补充快照机制，称为增量快照。增量快照依赖于 Debezium 机制将信号发送到 Debezium 连接器。增量快照基设计文档DDD-3: Incremental snapshotting。

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

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

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

6.增量快照过程

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

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

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

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

7.快照窗口

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

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

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

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

8.触发增量快照

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

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

要指定要包含在快照中的表，请提供列出表的数据集合数组，例如，
{“data-collections”: [“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 (_''_, _''_, '{"data-collections": ["__","__"],"type":"__"}');
1

例如，

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

命令中的id、type、data参数的值对应信令表的字段。

下表描述了这些参数：

表 3. 向信令表发送增量快照信号的 SQL 命令字段说明

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

{
    "before":null,
    "after": {
        "pk":"1",
        "value":"New data"
    },
    "source": {
        ...
        "snapshot":"incremental" 
    },
    "op":"r", 
    "ts_ms":"1620393591654",
    "transaction":null
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

注意：

• 用于 PostgreSQL 的 Debezium 连接器不支持在运行增量快照时更改架构。如果在增量快照开始之前但在发送信号之后执行架构更改，配置选项 database.autosave 设置为conservative以正确处理架构更改。

9.流式传输更改

PostgreSQL 连接器通常花费大部分时间从它所连接的 PostgreSQL 服务器流式传输更改。这种机制依赖于 PostgreSQL 的复制协议。该协议使客户端能够在服务器的事务日志中的某些位置提交更改时从服务器接收更改，这些位置称为日志序列号 (LSN)。

每当服务器提交事务时，一个单独的服务器进程都会调用逻辑解码插件的回调函数。此函数处理来自事务的更改，将它们转换为特定格式（在 Debezium 插件的情况下为 Protobuf 或 JSON）并将它们写入输出流，然后可由客户端使用。

Debezium PostgreSQL 连接器充当 PostgreSQL 客户端。当连接器收到更改时，它会将事件转换为 Debezium 创建、更新或删除事件，包括事件的 LSN。 PostgreSQL 连接器将记录中的这些更改事件转发到在同一进程中运行的 Kafka Connect 框架。 Kafka Connect 进程将更改事件记录按照生成它们的顺序异步写入相应的 Kafka 主题。

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

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

注意：

PostgreSQL 连接器检索模式信息作为逻辑解码插件发送的事件的一部分。但是，连接器不会检索有关哪些列构成主键的信息。连接器从 JDBC 元数据（侧通道）中获取此信息。如果表的主键定义发生变化（通过添加、删除或重命名主键列），则有一小段时间来自JDBC的主键信息与逻辑解码插件生成的更改事件不同步.在这个微小的时期内，可能会创建具有不一致密钥结构的消息。为防止这种不一致，请按如下方式更新主键结构：

• 将数据库或应用程序置于只读模式。
• 让 Debezium 处理所有剩余的事件。
• 停止Debezium。
• 更新相关表中的主键定义。
• 将数据库或应用程序置于读/写模式。
• 重新启动 Debezium。

10.PostgreSQL 10+ 逻辑解码支持 (pgoutput)

从 PostgreSQL 10+ 开始，有一种逻辑复制流模式，称为 pgoutput，它被 PostgreSQL 原生支持。这意味着 Debezium PostgreSQL 连接器可以使用该复制流，而无需额外的插件。这对于不支持或不允许安装插件的环境特别有价值。

11.Topic names

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

• serverName.schemaName.tableName

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

• 服务器名称：连接器的逻辑名称，由 database.server.name 配置属性指定。
• 模式名称：发生更改事件的数据库模式的名称。
• 表名：发生更改事件的数据库表的名称。

例如，假设fulfillment 是连接器配置中的逻辑服务器名称，该连接器正在捕获PostgreSQL 安装中的更改，postgres 数据库inventory模式包含四个表：products、products_on_hand、customers 和orders。连接器会将记录流式传输到这四个 Kafka 主题：

• fulfillment.inventory.products
• fulfillment.inventory.products_on_hand
• fulfillment.inventory.customers
• fulfillment.inventory.orders

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

• fulfillment.public.products
• fulfillment.public.products_on_hand
• fulfillment.public.customers
• fulfillment.public.orders

如果默认主题名称不符合您的要求，您可以配置自定义主题名称。要配置自定义主题名称，请在逻辑主题路由 SMT 中指定正则表达式。

12.元信息

除了数据库更改事件之外，PostgreSQL 连接器生成的每条记录都包含一些元数据。元数据包括事件在服务器上发生的位置、源分区的名称以及事件应该去的 Kafka 主题和分区的名称，例如：

"sourcePartition": {
     "server": "fulfillment"
 },
 "sourceOffset": {
     "lsn": "24023128",
     "txId": "555",
     "ts_ms": "1482918357011"
 },
 "kafkaPartition": null
1
2
3
4
5
6
7
8
9

• sourcePartition：始终默认为 database.server.name 连接器配置属性的设置。

• sourceOffset：包含有关发生事件的服务器位置的信息：

• lsn：表示事务日志中的 PostgreSQL 日志序列号或偏移量。

• txId：表示导致事件的服务器事务的标识符。

• ts_ms：以自纪元以​​来的毫秒数的形式表示事务提交的服务器时间。

• kafkaPartition：设置为 null 表示连接器不使用特定的 Kafka 分区。 PostgreSQL 连接器仅使用一个 Kafka Connect 分区，并将生成的事件放入一个 Kafka 分区。

13.事务元数据

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

注意：

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

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

• status：BEGIN or END
• id：唯一交易标识符的字符串表示
• event_count (for END events)：事务发出的事件总数
• data_collections（用于 END 事件）：data_collection 和 event_count 的数组，提供源自给定数据集合的更改发出的事件数

例子：

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

{
  "status": "END",
  "id": "571",
  "event_count": 2,
  "data_collections": [
    {
      "data_collection": "s1.a",
      "event_count": 1
    },
    {
      "data_collection": "s2.a",
      "event_count": 1
    }
  ]
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

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

14.丰富更改数据事件

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

• id：唯一交易标识符的字符串表示
• total_order：事件在事务生成的所有事件中的绝对位置
• data_collection_order：事件在事务发出的所有事件中的每个数据收集位置

以下是消息的示例：

{
  "before": null,
  "after": {
    "pk": "2",
    "aa": "1"
  },
  "source": {
   ...
  },
  "op": "c",
  "ts_ms": "1580390884335",
  "transaction": {
    "id": "571",
    "total_order": "1",
    "data_collection_order": "1"
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

三、数据更改事件

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

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

以下骨架 JSON 显示了更改事件的基本四个部分。如果您使用 JSON 转换器并将其配置为生成所有四个基本更改事件部分，则更改事件具有以下结构：

{
 "schema": { 
   ...
  },
 "payload": { 
   ...
 },
 "schema": { 
   ...
 },
 "payload": { 
   ...
 },
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

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

默认情况下，连接器流将事件记录更改为名称与事件源表相同的主题。

从 Kafka 0.10 开始，Kafka 可以选择使用创建消息（由生产者记录）或由 Kafka 写入日志的时间戳记录事件键和值。

PostgreSQL 连接器确保所有 Kafka Connect 模式名称都遵循 Avro 模式名称格式。这意味着逻辑服务器名称必须以拉丁字母或下划线开头，即 a-z、A-Z 或 _。逻辑服务器名称中的每个剩余字符以及模式和表名称中的每个字符都必须是拉丁字母、数字或下划线，即 a-z、A-Z、0-9 或 _。如果存在无效字符，则将其替换为下划线字符。

如果逻辑服务器名称、模式名称或表名称包含无效字符，并且将名称彼此区分开来的唯一字符无效并因此替换为下划线，这可能会导致意外冲突。

四、事件

REPLICA IDENTITY 是特定于 PostgreSQL 的表级设置，用于确定逻辑解码插件可用于 UPDATE 和 DELETE 事件的信息量。更具体地说，每当发生 UPDATE 或 DELETE 事件时，REPLICA IDENTITY 的设置控制了哪些信息（如果有）可用于所涉及的表列的先前值。

• DEFAULT：

• 如果该表具有主键，则默认行为是 UPDATE 和 DELETE 事件包含表的主键列的先前值。对于 UPDATE 事件，仅存在具有更改值的主键列。

• 如果表没有主键，则连接器不会为该表发出 UPDATE 或 DELETE 事件。对于没有主键的表，连接器仅发出创建事件。通常，没有主键的表用于将消息附加到表的末尾，这意味着 UPDATE 和 DELETE 事件没有用处。

• NOTHING：为 UPDATE 和 DELETE 操作发出的事件不包含有关任何表列的先前值的任何信息。

• FULL：为 UPDATE 和 DELETE 操作发出的事件包含表中所有列的先前值。

• INDEX：index-name - 为 UPDATE 和 DELETE 操作发出的事件包含指定索引中包含的列的先前值。 UPDATE 事件还包含具有更新值的索引列。

1.创建事件

{
    "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": "PostgreSQL_server.inventory.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": "PostgreSQL_server.inventory.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": "int64",
                        "optional": true,
                        "field": "txId"
                    },
                    {
                        "type": "int64",
                        "optional": true,
                        "field": "lsn"
                    },
                    {
                        "type": "int64",
                        "optional": true,
                        "field": "xmin"
                    }
                ],
                "optional": false,
                "name": "io.debezium.connector.postgresql.Source", 3
                "field": "source"
            },
            {
                "type": "string",
                "optional": false,
                "field": "op"
            },
            {
                "type": "int64",
                "optional": true,
                "field": "ts_ms"
            }
        ],
        "optional": false,
        "name": "PostgreSQL_server.inventory.customers.Envelope" 4
    },
    "payload": { 5
        "before": null, 6
        "after": { 7
            "id": 1,
            "first_name": "Anne",
            "last_name": "Kretchmar",
            "email": "annek@noanswer.org"
        },
        "source": { 8
            "version": "1.9.5.Final",
            "connector": "postgresql",
            "name": "PostgreSQL_server",
            "ts_ms": 1559033904863,
            "snapshot": true,
            "db": "postgres",
            "sequence": "[\"24023119\",\"24023128\"]"
            "schema": "public",
            "table": "customers",
            "txId": 555,
            "lsn": 24023128,
            "xmin": null
        },
        "op": "c", 9
        "ts_ms": 1559033904863 10
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164

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

2.更新事件

{
    "schema": { ... },
    "payload": {
        "before": { (1)
            "id": 1
        },
        "after": { (2)
            "id": 1,
            "first_name": "Anne Marie",
            "last_name": "Kretchmar",
            "email": "annek@noanswer.org"
        },
        "source": { (3)
            "version": "1.9.5.Final",
            "connector": "postgresql",
            "name": "PostgreSQL_server",
            "ts_ms": 1559033904863,
            "snapshot": false,
            "db": "postgres",
            "schema": "public",
            "table": "customers",
            "txId": 556,
            "lsn": 24023128,
            "xmin": null
        },
        "op": "u", (4)
        "ts_ms": 1465584025523  (5)
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

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

主键更新

更改行的主键字段的 UPDATE 操作称为主键更改。对于主键更改，连接器不是发送 UPDATE 事件记录，而是发送旧键的 DELETE 事件记录和新（更新的）键的 CREATE 事件记录。这些事件具有通常的结构和内容，此外，每个事件都有一个与主键更改相关的消息头：

• DELETE 事件记录具有 __debezium.newkey 作为消息头。此标头的值是更新行的新主键。
• CREATE 事件记录具有 __debezium.oldkey 作为消息头。此标头的值是更新行所具有的先前（旧）主键。

3.删除事件

{
    "schema": { ... },
    "payload": {
        "before": { 
            "id": 1
        },
        "after": null, 
        "source": { 
            "version": "1.9.5.Final",
            "connector": "postgresql",
            "name": "PostgreSQL_server",
            "ts_ms": 1559033904863,
            "snapshot": false,
            "db": "postgres",
            "schema": "public",
            "table": "customers",
            "txId": 556,
            "lsn": 46523128,
            "xmin": null
        },
        "op": "d", 
        "ts_ms": 1465581902461 
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

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

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

墓碑事件
当删除一行时，删除事件值仍然适用于日志压缩，因为 Kafka 可以删除所有具有相同键的早期消息。但是，要让 Kafka 删除具有相同键的所有消息，消息值必须为空。为了使这成为可能，PostgreSQL 连接器跟随一个带有特殊 tombstone 事件的删除事件，该事件具有相同的键但为空值。

4.截断事件

截断更改事件表示表已被截断。在这种情况下，消息键为空，消息值如下所示：

{
    "schema": { ... },
    "payload": {
        "source": { 
            "version": "1.9.5.Final",
            "connector": "postgresql",
            "name": "PostgreSQL_server",
            "ts_ms": 1559033904863,
            "snapshot": false,
            "db": "postgres",
            "schema": "public",
            "table": "customers",
            "txId": 556,
            "lsn": 46523128,
            "xmin": null
        },
        "op": "t", 
        "ts_ms": 1559033904961 
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

表 9. 截断事件值字段的描述

如果单个 TRUNCATE 语句适用于多个表，则会发出每个截断表的一个截断更改事件记录。

请注意，由于截断事件表示对整个表所做的更改并且没有消息键，除非您使用单个分区处理主题，否则与表相关的更改事件没有顺序保证（创建，更新等）并截断该表的事件。例如，当从不同分区读取这些事件时，消费者可能仅在该表的截断事件之后接收更新事件。

5.消息事件

此事件类型仅通过 Postgres 14+ 上的 pgoutput 插件支持。
消息事件表明通用逻辑解码消息已直接插入 WAL，通常使用 pg_logical_emit_message 函数。在这种情况下，消息键是一个结构体，具有一个名为 prefix 的字段，带有插入消息时指定的前缀。对于事务性消息，消息值如下所示：

{
    "schema": { ... },
    "payload": {
        "source": { 
            "version": "1.9.5.Final",
            "connector": "postgresql",
            "name": "PostgreSQL_server",
            "ts_ms": 1559033904863,
            "snapshot": false,
            "db": "postgres",
            "schema": "",
            "table": "",
            "txId": 556,
            "lsn": 46523128,
            "xmin": null
        },
        "op": "m", 
        "ts_ms": 1559033904961, 
        "message": { 
            "prefix": "foo",
            "content": "Ymfy"
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

与其他事件类型不同，非事务性消息不会有任何关联的 BEGIN 或 END 事务事件。对于非事务性消息，消息值如下所示：

{
    "schema": { ... },
    "payload": {
        "source": { 
            "version": "1.9.5.Final",
            "connector": "postgresql",
            "name": "PostgreSQL_server",
            "ts_ms": 1559033904863,
            "snapshot": false,
            "db": "postgres",
            "schema": "",
            "table": "",
            "lsn": 46523128,
            "xmin": null
        },
        "op": "m", 
        "ts_ms": 1559033904961 
        "message": { 
            "prefix": "foo",
            "content": "Ymfy"
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

表 10. 消息事件值字段说明

五、数据类型映射

PostgreSQL 连接器表示对具有事件的行的更改，这些事件的结构类似于该行所在的表。该事件包含每个列值的字段。该值在事件中的表示方式取决于列的 PostgreSQL 数据类型。以下部分描述了连接器如何将 PostgreSQL 数据类型映射到事件字段中的文字类型和语义类型。

• 文字类型描述了如何使用 Kafka Connect 模式类型从字面上表示值：INT8、INT16、INT32、INT64、FLOAT32、FLOAT64、BOOLEAN、STRING、BYTES、ARRAY、MAP 和 STRUCT。
• 语义类型描述了 Kafka Connect 模式如何使用字段的 Kafka Connect 模式的名称来捕获字段的含义。

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

1.基本类型

下表描述了连接器如何映射基本类型。

表 11. PostgreSQL 基本数据类型的映射

2.时间类型

除了包含时区信息的 PostgreSQL 的 TIMESTAMPTZ 和 TIMETZ 数据类型之外，时间类型的映射方式取决于 time.precision.mode 连接器配置属性的值。以下部分描述了这些映射：

• time.precision.mode=adaptive
• time.precision.mode=adaptive_time_microseconds
• time.precision.mode=connect

time.precision.mode=adaptive

当 time.precision.mode 属性设置为默认值 Adaptive 时，连接器根据列的数据类型定义确定文字类型和语义类型。这可确保事件准确地表示数据库中的值。

表 12. time.precision.mode 自适应时的映射

time.precision.mode=adaptive_time_microseconds
当 time.precision.mode 配置属性设置为 Adaptive_time_microseconds 时，连接器会根据列的数据类型定义确定时间类型的文字类型和语义类型。这可确保事件准确地表示数据库中的值，但所有 TIME 字段都被捕获为微秒。

表 13. time.precision.mode 为 Adaptive_time_microseconds 时的映射

time.precision.mode=连接
当 time.precision.mode 配置属性设置为 connect 时，连接器使用 Kafka Connect 逻辑类型。当消费者只能处理内置的 Kafka Connect 逻辑类型并且无法处理可变精度的时间值时，这可能很有用。但是，由于 PostgreSQL 支持微秒精度，当数据库列的小数秒精度值大于 3 时，具有连接时间精度模式的连接器生成的事件会导致精度损失。

表 14. time.precision.mode 连接时的映射

3.时间戳类型

TIMESTAMP 类型表示没有时区信息的时间戳。此类列将转换为基于 UTC 的等效 Kafka Connect 值。例如，当 time.precision.mode 未设置为连接时，TIMESTAMP 值“2018-06-20 15:13:16.945104”由值为“1529507596945104”的 io.debezium.time.MicroTimestamp 表示。

运行 Kafka Connect 和 Debezium 的 JVM 的时区不影响此转换。

PostgreSQL 支持在 TIMESTAMP 列中使用 +/-infinite 值。这些特殊值在正无穷大的情况下转换为值为 9223372036825200000 的时间戳，在负无穷大的情况下转换为 -9223372036832400000。此行为模仿 PostgreSQL JDBC 驱动程序的标准行为 。

4.小数类型

PostgreSQL 连接器配置属性 decimal.handling.mode 的设置决定了连接器如何映射十进制类型。

当 decimal.handling.mode 属性设置为precise时，连接器对所有 DECIMAL、NUMERIC 和 MONEY 列使用 Kafka Connect org.apache.kafka.connect.data.Decimal 逻辑类型。这是默认模式。

表 15. decimal.handling.mode 精确时的映射

这条规则有一个例外。当 NUMERIC 或 DECIMAL 类型在没有比例限制的情况下使用时，来自数据库的值对每个值都有不同的（可变）比例。在这种情况下，连接器使用 io.debezium.data.VariableScaleDecimal，其中包含传输值的值和小数位数。

表 16. 没有比例约束时 DECIMAL 和 NUMERIC 类型的映射

当 decimal.handling.mode 属性设置为 double 时，连接器将所有 DECIMAL、NUMERIC 和 MONEY 值表示为 Java double 值，并如下表所示对它们进行编码。

表 17. decimal.handling.mode 为 double 时的映射

decimal.handling.mode 配置属性的最后一个可能设置是字符串。在这种情况下，连接器将 DECIMAL、NUMERIC 和 MONEY 值表示为其格式化的字符串表示形式，并如下表所示对它们进行编码。
表18. decimal.handling.mode 为字符串时的映射

当 decimal.handling.mode 的设置为 string 或 double 时，PostgreSQL 支持将 NaN（不是数字）作为特殊值存储在 DECIMAL/NUMERIC 值中。在这种情况下，连接器将 NaN 编码为 Double.NaN 或字符串常量 NAN。

六、默认值

如果为数据库模式中的列指定了默认值，PostgreSQL 连接器将尽可能尝试将此值传播到 Kafka 模式。支持最常见的数据类型，包括：

• BOOLEAN
• Numeric types (INT, FLOAT, NUMERIC, etc.)
• Text types (CHAR, VARCHAR, TEXT, etc.)
• Temporal types (DATE, TIME, INTERVAL, TIMESTAMP, TIMESTAMPTZ)
• JSON, JSONB, XML
• UUID

请注意，对于时间类型，默认值的解析由 PostgreSQL 库提供；因此，通常 PostgreSQL 支持的任何字符串表示形式也应该被连接器支持。

在默认值由函数生成而不是直接在行内指定的情况下，连接器将改为导出给定数据类型的等效值 0。这些值包括：

• 布尔值为 FALSE
• 0 具有适当的精度，用于数字类型
• 文本/XML 类型的空字符串
• {} 用于 JSON 类型
• 1970-01-01 适用于 DATE、TIMESTAMP、TIMESTAMPTZ 类型
• 00:00 时间
• INTERVAL 的 EPOCH
• 00000000-0000-0000-0000-000000000000 用于 UUID

此支持目前仅扩展到函数的显式使用。例如，括号支持 CURRENT_TIMESTAMP(6)，但不支持 CURRENT_TIMESTAMP。

七、设置 Postgres

1.配置 PostgreSQL 服务器

如果你使用的是除pgoutput之外的逻辑解码插件，安装后，配置PostgreSQL服务器如下：

• 要在启动时加载插件，请将以下内容添加到 postgresql.conf 文件中：

# MODULES
shared_preload_libraries = 'decoderbufs' 
1
2

• 指示服务器在启动时加载decoderbufs逻辑解码插件（插件名称在Protobuf make文件中设置）。

• 要配置复制槽而不考虑使用的解码器，请在 postgresql.conf 文件中指定以下内容：

# REPLICATION
wal_level = logical   
1
2

• 指示服务器对预写日志使用逻辑解码。

需要在使用 Debezium 时设置其他 PostgreSQL 流式复制参数。示例包括 max_wal_senders 和 max_replication_slots 用于增加可以同时访问发送服务器的连接器数量，以及 wal_keep_size 用于限制复制槽将保留的最大 WAL 大小。

Debezium 使用 PostgreSQL 的逻辑解码，它使用复制槽。即使在 Debezium 中断期间，复制槽也保证保留 Debezium 所需的所有 WAL 段。出于这个原因，密切监视复制槽非常重要，以避免过多的磁盘消耗和其他可能发生的情况，例如如果复制槽长时间未使用，则可能发生目录膨胀。

如果正在使用除 on 以外的 synchronous_commit 设置，建议将 wal_writer_delay 设置为 10 毫秒等值，以实现更改事件的低延迟。否则，将应用其默认值，这会增加大约 200 毫秒的延迟。

2.设置权限

设置 PostgreSQL 服务器以运行 Debezium 连接器需要可以执行复制的数据库用户。复制只能由具有适当权限的数据库用户执行，并且只能用于配置数量的主机。

尽管默认情况下，超级用户具有必要的 REPLICATION 和 LOGIN 角色，如安全中所述，但最好不要为 Debezium 复制用户提供提升的权限。相反，创建一个具有最低要求权限的 Debezium 用户。

先决条件

• PostgreSQL 管理权限。

程序

• 要为用户提供复制权限，请定义至少具有 REPLICATION 和 LOGIN 权限的 PostgreSQL 角色，然后将该角色授予用户。例如：

CREATE ROLE <name> REPLICATION LOGIN;
1

设置权限以启用 Debezium 在使用 pgoutput 时创建 PostgreSQL 发布

如果使用 pgoutput 作为逻辑解码插件，则 Debezium 必须以具有特定权限的用户身份在数据库中操作。

Debezium 从为表创建的发布中流式传输 PostgreSQL 源表的更改事件。发布包含从一个或多个表生成的一组过滤的更改事件。根据发布规范过滤每个发布中的数据。该规范可以由 PostgreSQL 数据库管理员或 Debezium 连接器创建。要允许 Debezium PostgreSQL 连接器创建发布并指定要复制到其中的数据，连接器必须在数据库中以特定权限运行。

要让 Debezium 创建 PostgreSQL 发布，它必须以具有以下权限的用户身份运行：

• 将表添加到发布的数据库中的复制权限。
• 数据库上的 CREATE 权限以添加发布。
• 表上的 SELECT 权限以复制初始表数据。表所有者自动拥有表的 SELECT 权限。

要将表添加到发布，用户必须是表的所有者。但是由于源表已经存在，您需要一种机制来与原始所有者共享所有权。要启用共享所有权，您需要创建一个 PostgreSQL 复制组，然后将现有表所有者和复制用户添加到该组。

程序

• 创建复制组。

CREATE ROLE <replication_group>;
1

• 将表的原始所有者添加到组中。

GRANT REPLICATION_GROUP TO <original_owner>;
1

• 将 Debezium 复制用户添加到该组。

GRANT REPLICATION_GROUP TO <replication_user>;
1

• 将表的所有权转移到 。

ALTER TABLE <table_name> OWNER TO REPLICATION_GROUP;
1

为了让 Debezium 指定捕获配置，publication.autocreate.mode 的值必须设置为过滤。

3.配置 PostgreSQL 以允许使用 Debezium 连接器主机进行复制

要使 Debezium 能够复制 PostgreSQL 数据，必须将数据库配置为允许与运行 PostgreSQL 连接器的主机进行复制。要指定允许与数据库一起复制的客户端，请将条目添加到基于 PostgreSQL 主机的身份验证文件 pg_hba.conf。

程序

• 向 pg_hba.conf 文件添加条目以指定可以与数据库主机复制的 Debezium 连接器主机。例如，

pg_hba.conf file example:

local   replication     <youruser>                          trust   
host    replication     <youruser>  127.0.0.1/32            trust   
host    replication     <youruser>  ::1/128                 trust   
1
2
3

• 指示服务器允许在本地（即在服务器计算机上）复制 。
• 指示服务器允许本地主机上的 使用 IPV4 接收复制更改。
• 指示服务器允许本地主机上的 使用 IPV6 接收复制更改。

八、debezium connector配置参数详解

1.debezium connector配置

以下是 PostgreSQL 连接器的配置示例，该连接器在 192.168.99.100 的端口 5432 上连接到 PostgreSQL 服务器，其逻辑名称为履行。通常，您通过设置可用于连接器的配置属性，在 JSON 文件中配置 Debezium PostgreSQL 连接器。

{
  "name": "fulfillment-connector",  
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector", 
    "database.hostname": "192.168.99.100", 
    "database.port": "5432", 
    "database.user": "postgres", 
    "database.password": "postgres", 
    "database.dbname" : "postgres", 
    "database.server.name": "fulfillment", 
    "table.include.list": "public.inventory" 

  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

• 向 Kafka Connect 服务注册时的连接器名称。
• 此 PostgreSQL 连接器类的名称。
• PostgreSQL 服务器的地址。
• PostgreSQL 服务器的端口号。
• 具有所需权限的 PostgreSQL 用户的名称。
• 具有所需权限的 PostgreSQL 用户的密码。
• 要连接的 PostgreSQL 数据库的名称
• PostgreSQL 服务器/集群的逻辑名称，形成一个命名空间，用于连接器写入的所有 Kafka 主题的名称、Kafka Connect 模式名称以及 Avro 转换器运行时相应 Avro 模式的命名空间用过的。
• 此连接器将监视的此服务器托管的所有表的列表。这是可选的，还有其他属性用于列出要从监视中包含或排除的模式和表。

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

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

添加连接器配置
要运行 Debezium PostgreSQL 连接器，请创建一个连接器配置并将该配置添加到您的 Kafka Connect 集群。

先决条件

• PostgreSQL 被配置为支持逻辑复制。
• 已安装逻辑解码插件。
• PostgreSQL 连接器已安装。

程序

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

结果

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

2.连接器配置属性

Debezium PostgreSQL 连接器具有许多配置属性，您可以使用这些属性为您的应用程序实现正确的连接器行为。许多属性都有默认值。有关属性的信息组织如下：

• 必需的配置属性
• 高级配置属性
• 传递配置属性

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

表 22. 所需的连接器配置属性

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

表 23. 高级连接器配置属性

九、postgresql数据库Debezium连接器的监控指标

除了 Zookeeper、Kafka 和 Kafka Connect 提供的对 JMX 指标的内置支持之外，Debezium PostgreSQL 连接器还提供两种类型的指标。

• 快照指标在执行快照时提供有关连接器操作的信息。
• 当连接器捕获更改和流式处理更改事件记录时，流式处理指标提供有关连接器操作的信息。

安装jmx导出器监控postgresql数据库debezium连接器的指标可以参考下面这篇博客：

• Debezium系列之：安装jmx导出器监控debezium指标

1.Snapshot metrics

• MBean 是 debezium.postgres:type=connector-metrics,context=snapshot,server=。
• 除非快照操作处于活动状态，或者自上次连接器启动以来已发生快照，否则不会公开快照指标。

下表列出了可用的快照指标。

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

2.Streaming metrics

MBean 是 debezium.postgres:type=connector-metrics,context=streaming,server=。

下表列出了可用的流式指标。

十、debezium同步postgresql数据库数据实际应用详细步骤

安装Debezium集群的详细步骤可以参考博主下面这篇博客文章：

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

debezium同步postgresql数据库数据实际应用详细步骤请参考博主下面这篇博客

• Debezium系列之：debezium同步postgresql数据库数据到kafka集群