Debezium PostgreSQL 连接器

要使用 Debezium 连接器从 PostgreSQL 数据库流式传输更改，连接器必须在数据库中拥有特定权限。虽然授予必要权限的一种方法是为用户提供superuser权限，但这可能会使您的 PostgreSQL 数据暴露给未经授权的访问。与其向 Debezium 用户授予过多的权限，不如创建一个专用的 Debezium 复制用户，并向其授予特定权限。

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

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

以下步骤描述了连接器在初始快照期间执行的默认步骤。您可以将snapshot.mode连接器配置属性设置为initial以外的值来更改此行为。

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

By default, a connector runs an initial snapshot operation only after it starts for the first time. Following this initial snapshot, under normal circumstances, the connector does not repeat the snapshot process. Any future change event data that the connector captures comes in through the streaming process only. (默认情况下，连接器仅在首次启动后运行初始快照操作。在初始快照之后，在正常情况下，连接器不会重复快照过程。连接器捕获的任何未来变更事件数据仅通过流式传输过程进入。)

However, in some situations the data that the connector obtained during the initial snapshot might become stale, lost, or incomplete. To provide a mechanism for recapturing table data, Debezium includes an option to perform ad hoc snapshots. You might want to perform an ad hoc snapshot after any of the following changes occur in your Debezium environment (但是，在某些情况下，连接器在初始快照期间获取的数据可能会过时、丢失或不完整。为了提供重新捕获表数据的机制，Debezium 提供了一个执行即席快照的选项。您可能希望在 Debezium 环境中发生以下任何更改后执行即席快照：)

You can re-run a snapshot for a table for which you previously captured a snapshot by initiating a so-called ad hoc snapshot. Ad hoc snapshots require the use of signaling tables. You initiate an ad hoc snapshot by sending a signal request to the Debezium signaling table. (您可以通过启动所谓的即席快照来重新运行之前捕获了快照的表的快照。即席快照需要使用信号表。通过向 Debezium 信号表发送信号请求来启动即席快照。)

When you initiate an ad hoc snapshot of an existing table, the connector appends content to the topic that already exists for the table. If a previously existing topic was removed, Debezium can create a topic automatically if automatic topic creation is enabled. (当您启动现有表的即席快照时，连接器会将内容追加到该表已存在的主题中。如果之前存在的主题已被删除，并且启用了自动主题创建，则 Debezium 可以自动创建主题。)

Ad hoc snapshot signals specify the tables to include in the snapshot. The snapshot can capture the entire contents of the database, or capture only a subset of the tables in the database. Also, the snapshot can capture a subset of the contents of the table(s) in the database. (即席快照信号指定要包含在快照中的表。快照可以捕获数据库的全部内容，或仅捕获数据库中表的部分内容。此外，快照还可以捕获数据库中表的部分内容。)

You specify the tables to capture by sending an execute-snapshot message to the signaling table. Set the type of the execute-snapshot signal to incremental or blocking, and provide the names of the tables to include in the snapshot, as described in the following table (您可以通过向信号表发送 execute-snapshot 消息来指定要捕获的表。将 execute-snapshot 信号的类型设置为 incremental 或 blocking，并提供要在快照中包含的表名称，如下表所示：)

通过向信号表添加一个具有execute-snapshot信号类型的条目，或者将信号消息发送到 Kafka 信号主题，来启动即席增量快照。在连接器处理完消息后，它将开始快照操作。快照过程读取第一个和最后一个主键值，并将这些值用作每个表的开始和结束点。根据表中的条目数和配置的块大小，Debezium 将表分成块，并逐个连续地对每个块进行快照。

有关更多信息，请参阅增量快照。

You initiate an ad hoc blocking snapshot by adding an entry with the execute-snapshot signal type to the signaling table or signaling topic. After the connector processes the message, it begins the snapshot operation. The connector temporarily stops streaming, and then initiates a snapshot of the specified table, following the same process that it uses during an initial snapshot. After the snapshot completes, the connector resumes streaming. (您可以通过向信号表或信号主题添加具有 execute-snapshot 信号类型的条目来启动即席阻塞快照。在连接器处理消息后，它将开始快照操作。连接器将暂时停止流式传输，然后启动指定表的快照，遵循其在初始快照期间使用的相同过程。快照完成后，连接器将恢复流式传输。)

有关更多信息，请参阅阻塞快照。

To provide flexibility in managing snapshots, Debezium includes a supplementary snapshot mechanism, known as incremental snapshotting. Incremental snapshots rely on the Debezium mechanism for sending signals to a Debezium connector. Incremental snapshots are based on the DDD-3 design document. (为了在管理快照时提供灵活性，Debezium 提供了一个补充快照机制，称为增量快照。增量快照依赖于 Debezium 向 Debezium 连接器发送信号的机制。增量快照基于DDD-3 设计文档。)

在增量快照中，Debezium 不是一次性捕获数据库的完整状态（如初始快照），而是分阶段、以一系列可配置的块捕获每个表。您可以指定要快照的表以及每个块的大小。块大小决定了快照在每次从数据库获取操作期间收集的行数。增量快照的默认块大小为 1024 行。

As an incremental snapshot proceeds, Debezium uses watermarks to track its progress, maintaining a record of each table row that it captures. This phased approach to capturing data provides the following advantages over the standard initial snapshot process (随着增量快照的进行，Debezium 使用水位标记来跟踪其进度，并维护捕获的每个表行的记录。与标准的初始快照过程相比，这种分阶段的数据捕获方法提供了以下优势：)

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

As a snapshot proceeds, it’s likely that other processes continue to access the database, potentially modifying table records. To reflect such changes, INSERT, UPDATE, or DELETE operations are committed to the transaction log as per usual. Similarly, the ongoing Debezium streaming process continues to detect these change events and emits corresponding change event records to Kafka. (随着快照的进行，其他进程很可能会继续访问数据库，可能修改表记录。为了反映这些更改，INSERT、UPDATE 或 DELETE 操作照常提交到事务日志。同样，正在进行的 Debezium 流式传输过程会继续检测这些变更事件，并将相应的变更事件记录发送到 Kafka。)

In some cases, the UPDATE or DELETE events that the streaming process emits are received out of sequence. That is, the streaming process might emit an event that modifies a table row before the snapshot captures the chunk that contains the READ event for that row. When the snapshot eventually emits the corresponding READ event for the row, its value is already superseded. To ensure that incremental snapshot events that arrive out of sequence are processed in the correct logical order, Debezium employs a buffering scheme for resolving collisions. Only after collisions between the snapshot events and the streamed events are resolved does Debezium emit an event record to Kafka. (在某些情况下，流式传输过程发出的 UPDATE 或 DELETE 事件可能会乱序接收。也就是说，流式传输过程可能会在快照捕获包含该行 READ 事件的块之前发出修改表行的事件。当快照最终发出行的相应 READ 事件时，其值已过时。为了确保乱序到达的增量快照事件按正确的逻辑顺序处理，Debezium 采用缓冲方案来解决冲突。仅当快照事件与流式事件之间的冲突得到解决后，Debezium 才会向 Kafka 发送事件记录。)

To assist in resolving collisions between late-arriving READ events and streamed events that modify the same table row, Debezium employs a so-called snapshot window. The snapshot window demarcates the interval during which an incremental snapshot captures data for a specified table chunk. Before the snapshot window for a chunk opens, Debezium follows its usual behavior and emits events from the transaction log directly downstream to the target Kafka topic. But from the moment that the snapshot for a particular chunk opens, until it closes, Debezium performs a de-duplication step to resolve collisions between events that have the same primary key. (为了帮助解决延迟到达的 READ 事件与修改同一表行的流式事件之间的冲突，Debezium 采用所谓的快照窗口。快照窗口划定了增量快照捕获指定表块数据的间隔。在某个块的快照窗口打开之前，Debezium 会遵循其常规行为，将事务日志中的事件直接向下游发送到目标 Kafka 主题。但在某个块的快照打开到关闭的整个过程中，Debezium 会执行去重步骤以解决具有相同主键的事件之间的冲突。)

For each data collection, the Debezium emits two types of events, and stores the records for them both in a single destination Kafka topic. The snapshot records that it captures directly from a table are emitted as READ operations. Meanwhile, as users continue to update records in the data collection, and the transaction log is updated to reflect each commit, Debezium emits UPDATE or DELETE operations for each change. (对于每个数据集合，Debezium 会发出两种类型的事件，并将两者的记录存储在单个目标 Kafka 主题中。它直接从表中捕获的快照记录作为 READ 操作发出。同时，随着用户继续更新数据集合中的记录，并且事务日志更新以反映每次提交，Debezium 会为每个更改发出 UPDATE 或 DELETE 操作。)

As the snapshot window opens, and Debezium begins processing a snapshot chunk, it delivers snapshot records to a memory buffer. During the snapshot windows, the primary keys of the READ events in the buffer are compared to the primary keys of the incoming streamed events. If no match is found, the streamed event record is sent directly to Kafka. If Debezium detects a match, it discards the buffered READ event, and writes the streamed record to the destination topic, because the streamed event logically supersede the static snapshot event. After the snapshot window for the chunk closes, the buffer contains only READ events for which no related transaction log events exist. Debezium emits these remaining READ events to the table’s Kafka topic. (随着快照窗口的打开，Debezium 开始处理快照块，它会将快照记录传递到内存缓冲区。在快照窗口期间，缓冲区中 READ 事件的主键与传入的流式事件的主键进行比较。如果未找到匹配项，则将流式事件记录直接发送到 Kafka。如果 Debezium 检测到匹配，它将丢弃已缓冲的 READ 事件，并将流式记录写入目标主题，因为流式事件在逻辑上取代了静态快照事件。当块的快照窗口关闭后，缓冲区将仅包含没有相关事务日志事件的 READ 事件。Debezium 将这些剩余的 READ 事件发送到表的 Kafka 主题。)

The connector repeats the process for each snapshot chunk. (连接器对每个快照块重复此过程。)

Currently, you can use either of the following methods to initiate an incremental snapshot (当前，您可以使用以下任一方法启动增量快照：)

For more advanced uses, you can fine-tune control of the snapshot by implementing one of the following interfaces (对于更高级的用途，您可以实现以下接口之一来微调快照的控制：)

To provide more flexibility in managing snapshots, Debezium includes a supplementary ad hoc snapshot mechanism, known as a blocking snapshot. Blocking snapshots rely on the Debezium mechanism for sending signals to a Debezium connector. (为了在管理快照时提供更大的灵活性，Debezium 提供了一个补充的即席快照机制，称为阻塞快照。阻塞快照依赖于 Debezium 向 Debezium 连接器发送信号的机制。)

A blocking snapshot behaves just like an initial snapshot, except that you can trigger it at run time. (阻塞快照的行为与初始快照完全相同，只是您可以随时在运行时触发它。)

You might want to run a blocking snapshot rather than use the standard initial snapshot process in the following situations (您可能希望在以下情况下运行阻塞快照，而不是使用标准的初始快照过程：)

When you run a blocking snapshot, Debezium stops streaming, and then initiates a snapshot of the specified table, following the same process that it uses during an initial snapshot. After the snapshot completes, the streaming is resumed. (运行阻塞快照时，Debezium 会停止流式传输，然后启动指定表的快照，遵循其在初始快照期间使用的相同过程。快照完成后，流式传输将恢复。)

You can set the following properties in the data component of a signal (您可以在信号的 data 组件中设置以下属性：)

For example (例如：)

A delay might exist between the time that you send the signal to trigger the snapshot, and the time when streaming stops and the snapshot starts. As a result of this delay, after the snapshot completes, the connector might emit some event records that duplicate records captured by the snapshot. (从您发送触发快照的信号到流式传输停止并开始快照之间可能存在延迟。由于此延迟，快照完成后，连接器可能会发出一些重复快照捕获记录的事件记录。)

PostgreSQL 连接器通常花费大部分时间从其连接的 PostgreSQL 服务器流式传输更改。此机制依赖于PostgreSQL 的复制协议。此协议允许客户端在事务日志中某些位置（称为日志序列号 LSN）提交更改时接收来自服务器的更改。

每当服务器提交事务时，一个单独的服务器进程会调用逻辑解码插件中的回调函数。该函数处理事务中的更改，将其转换为特定格式（Protobuf 或 JSON，如果是 Debezium 插件），并将其写入输出流，然后客户端可以消费该输出流。

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

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

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

从 PostgreSQL 10+ 开始，有一个称为 pgoutput 的逻辑复制流模式，由 PostgreSQL 原生支持。这意味着 Debezium PostgreSQL 连接器可以在不需要额外插件的情况下消耗该复制流。这对于不允许安装插件的环境尤其有用。

有关更多信息，请参阅设置 PostgreSQL。

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

topicPrefix.schemaName.tableName

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

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

现在假设这些表不属于特定模式，而是创建在默认的 public PostgreSQL 模式中。Kafka 主题的名称将是：

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

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

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

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

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

启用事务元数据后，数据消息 Envelope 将使用新的 transaction 字段进行丰富。该字段以字段的复合形式提供关于每个事件的信息：

以下是消息示例

对于给定的表，更改事件的键的结构包含一个字段，该字段对应于事件创建时表中主键中的每个列。或者，如果表的 REPLICA IDENTITY 设置为 FULL 或 USING INDEX，则有一个字段对应于每个唯一键约束。

考虑一个定义在 public 数据库模式中的 customers 表，以及该表的更改事件键的示例。

如果 topic.prefix 连接器配置属性的值为 PostgreSQL_server，则在此定义下，customers 表的所有更改事件都具有相同的键结构，该结构在 JSON 中如下所示：

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

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

对于此表更改的更改事件的值部分根据 REPLICA IDENTITY 设置和事件的操作而变化。

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

REPLICA IDENTITY 有 4 个可能的值：

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

对于样本 customers 表中的更新，更改事件的值具有与该表的创建事件相同的模式。同样，事件值

的有效负载也具有相同的结构。但是，在更新事件中，事件值有效负载包含不同的值。以下是一个更改事件值在连接器为 customers 表中的更新生成的事件中的示例：

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

删除更改事件中的值具有与

同一表的创建和更新事件相同的 schema 部分。对于样本 customers 表，删除事件中的 payload 部分如下所示：

*delete* 更改事件记录为使用者提供了处理该行删除所需的信息。

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

当行被删除时，delete 事件值仍然可以与日志压缩配合使用，因为 Kafka 可以删除具有相同键的所有先前消息。但是，要使 Kafka 能够删除具有相同键的所有消息，消息值必须为null。为了实现这一点，PostgreSQL 连接器会在delete事件后跟一个特殊的墓碑事件，该事件具有相同的键但值为null。

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

如果单个 TRUNCATE 操作影响多个表，连接器会为每个被截断的表发出一个截断更改事件记录。

如果您不希望连接器捕获truncate事件，请使用skipped.operations选项将其过滤掉。

message 事件表示一个通用逻辑解码消息已直接插入到 WAL 中，通常使用 pg_logical_emit_message 函数。消息键是一个 Struct，其中包含一个名为 prefix 的字段，该字段承载插入消息时指定的名称。消息值在事务性消息中的样子如下：

与其他事件类型不同，非事务性消息将没有任何关联的 BEGIN 或 END 事务事件。消息值在非事务性消息中的样子如下：

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

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

当 time.precision.mode 属性设置为adaptive（默认值）时，连接器根据列的数据类型定义确定字面类型和语义类型。这确保了事件精确表示数据库中的值。

当 time.precision.mode 配置属性设置为 adaptive_time_microseconds 时，连接器根据列的数据类型定义确定时间类型的字面类型和语义类型。这确保了事件精确表示数据库中的值，只是所有 TIME 字段都捕获为微秒。

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

将 time.precision.mode 属性设置为 isostring，以将连接器配置为将时间值映射为 ISO-8601 格式的字符串，时间为 UTC。应用此设置时，连接器使用语义类型 io.debezium.time.IsoTimestamp、io.debezium.time.IsoTime 和 io.debezium.time.IsoDate 来映射时间戳、日期时间、日期和时间值。

将 time.precision.mode 属性设置为 microseconds，以将连接器配置为以微秒精度指定时间值。应用此设置时，连接器使用语义类型 io.debezium.time.MicroTime 和 io.debezium.time.MicroTimestamp 来映射时间戳、日期时间、时间值。

将 time.precision.mode 属性设置为 nanoseconds，以将连接器配置为以纳秒精度指定时间值。应用此设置时，连接器使用语义类型 io.debezium.time.NanoTime 和 io.debezium.time.NanoTimestamp（它们以纳秒精度存储值）来映射时间戳、日期时间、时间值。

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

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

PostgreSQL 支持在 TIMESTAMP 列中使用+/-infinite值。这些特殊值转换为时间戳，对于正无穷大值为 9223372036825200000，对于负无穷大值为 -9223372036832400000。此行为模仿了 PostgreSQL JDBC 驱动程序的标准行为。有关参考，请参阅org.postgresql.PGStatement接口。

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

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

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

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

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

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

PostgreSQL 连接器配置属性hstore.handling.mode的设置决定了连接器如何映射 HSTORE 值。

当 hstore.handling.mode 属性设置为 json（默认）时，连接器将 HSTORE 值表示为 JSON 值的字符串表示，并按如下表所示进行编码。当 hstore.handling.mode 属性设置为 map 时，连接器使用 MAP 模式类型来表示 HSTORE 值。

PostgreSQL 支持基于其他底层类型的用户定义类型。当使用此类列类型时，Debezium 会根据完整的类型层次结构公开列的表示。

PostgreSQL 具有可以存储 IPv4、IPv6 和 MAC 地址的数据类型。最好使用这些类型而不是纯文本类型来存储网络地址。网络地址类型提供输入错误检查和专用运算符及函数。

PostgreSQL 连接器支持所有 PostGIS 数据类型。

PostgreSQL 连接器支持所有 pgvector 扩展数据类型。

PostgreSQL 对页面大小有限制。这意味着大于 8 KB 的值需要使用TOAST 存储来存储。这会影响从数据库传入的复制消息。使用 TOAST 机制存储且未更改的值不包含在消息中，除非它们是表副本标识的一部分。Debezium 没有安全的方法可以直接从数据库中读取缺失的值，因为这可能会导致竞态条件。因此，Debezium 遵循以下规则来处理 toasted 值：

如果在数据库模式中为列指定了默认值，PostgreSQL 连接器将尝试在可能的情况下将此值传播到 Kafka 模式。大多数常见数据类型都支持，包括以下内容：

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

如果默认值是由函数生成而不是直接内联指定的，则连接器将导出给定数据类型的 0 的等效值。这些值包括：

此支持目前仅限于函数的显式使用。例如，CURRENT_TIMESTAMP(6) 在带括号时支持，但 CURRENT_TIMESTAMP 不支持。

从 PostgreSQL 9.4 开始，读取写前日志更改的唯一方法是安装逻辑解码输出插件。插件是用 C 编写的，经过编译并安装在运行 PostgreSQL 服务器的机器上。插件使用一些 PostgreSQL 特定的 API，如PostgreSQL 文档所述。

PostgreSQL 连接器与 Debezium 支持的逻辑解码插件之一配合使用，以在 Protobuf 格式或 pgoutput 格式接收来自数据库的更改事件。pgoutput 插件随 PostgreSQL 数据库一起提供。有关通过 decoderbufs 插件使用 Protobuf 的更多详细信息，请参阅插件文档，其中讨论了其要求、限制以及如何编译它。

为了简化，Debezium 还提供了一个基于上游 PostgreSQL 服务器镜像的容器镜像，在此基础上编译和安装插件。您可以使用此镜像作为安装所需详细步骤的示例。

对于所有情况，插件的行为并非完全相同。已确定以下差异：

所有最新的差异都跟踪在一个测试套件的 Java 类中。

如果您使用的逻辑解码插件 不是 pgoutput，请在安装后按以下方式配置 PostgreSQL 服务器。

根据您的需求，在使用 Debezium 时，您可能需要设置其他 PostgreSQL 流复制参数。例如，max_wal_senders 和 max_replication_slots 用于增加可以同时访问发送服务器的连接器数量，以及 wal_keep_size 用于限制复制槽将保留的最大 WAL 大小。有关配置流复制的更多信息，请参阅 PostgreSQL 文档。

Debezium 使用 PostgreSQL 的逻辑解码，该逻辑解码使用复制槽。即使在 Debezium 出现故障期间，复制槽也能保证保留 Debezium 所需的所有 WAL 段。因此，密切监控复制槽以避免磁盘占用过大以及复制槽长时间未使用的其他可能情况（例如目录膨胀）非常重要。有关更多信息，请参阅 PostgreSQL 流复制文档。

当连接到 Postgres 主服务器（即非只读副本）且 Postgres 版本为 17 或更高版本时，Debezium 将创建一个启用 故障转移 的复制槽。这意味着在发生故障时，Debezium 可以从提升为主服务器的副本继续读取更改，而不会丢失任何事件。这需要将槽的状态从主服务器同步到副本服务器；有关详细信息，请参阅 Postgres 文档。

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

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

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

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

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

有几种方法可以确定数据库的创建方式。通常，最好在设置连接器之前手动创建要捕获的表的数据库。但是，您可以配置环境以允许 Debezium 自动创建数据库，并指定要添加到其中的数据。

Debezium 使用包含列表和排除列表属性来指定数据如何插入到数据库中。有关允许 Debezium 创建数据库的选项的更多信息，请参阅 publication.autocreate.mode。

要使 Debezium 能够创建 PostgreSQL 数据库，它必须以具有以下权限的用户身份运行：

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

要使 Debezium 能够指定捕获配置，publication.autocreate.mode 的值必须设置为 filtered。

要使 Debezium 能够复制 PostgreSQL 数据，您必须配置数据库以允许与运行 PostgreSQL 连接器的主机进行复制。要指定允许复制的客户端，请将条目添加到基于主机的 PostgreSQL 身份验证文件 pg_hba.conf。有关 pg_hba.conf 文件的更多信息，请参阅 PostgreSQL 文档。

PostgreSQL 连接器可以与独立的 PostgreSQL 服务器或 PostgreSQL 服务器集群一起使用。

在某些情况下，WAL 文件占用的 PostgreSQL 磁盘空间可能会激增或超出正常比例。这种情况有几个可能的原因：

Debezium 使用复制槽从数据库流式传输更改。这些复制槽维护当前位置，形式为 LSN（日志序列号），它是 Debezium 连接器正在使用的 WAL 中的一个指针。这有助于 PostgreSQL 保持 WAL 可用，直到它被 Debezium 处理。单个复制槽只能由单个消费者或进程使用，因为不同的消费者可能具有不同的状态，并且可能需要来自不同位置的数据。

由于一个复制槽只能由一个连接器使用，因此为每个 Debezium 连接器创建一个唯一的复制槽至关重要。尽管当连接器不活动时，Postgres 可能允许其他连接器消耗复制槽，但这可能很危险，因为它可能导致数据丢失，因为一个槽只会发出每个更改一次 [更多信息]。

除了复制槽之外，在使用 pgoutput 插件时，Debezium 还使用数据库来流式传输事件。与复制槽类似，数据库是在数据库级别定义的，并为一组表定义。因此，除非连接器处理相同的表集，否则您将需要为每个连接器创建唯一的数据库。有关允许 Debezium 创建数据库的选项的更多信息，请参阅 publication.autocreate.mode。

有关如何为每个连接器设置唯一的复制槽名称和数据库名称，请参阅 slot.name 和 publication.name。

当您升级 Debezium 使用的 PostgreSQL 数据库时，您必须采取特定步骤来防止数据丢失并确保 Debezium 继续运行。通常，Debezium 对网络故障和其他中断引起的故障具有弹性。例如，当连接器监控的数据库服务器停止或崩溃时，在连接器重新建立与 PostgreSQL 服务器的通信后，它将继续从日志序列号（LSN）偏移量记录的最后位置读取。连接器从 Kafka Connect 偏移量主题检索有关最后记录偏移量的信息，并查询配置的 PostgreSQL 复制槽以获取具有相同值的日志序列号（LSN）。

要使连接器能够启动并捕获 PostgreSQL 数据库的更改事件，必须存在一个复制槽。但是，作为 PostgreSQL 升级过程的一部分，会删除复制槽，并且在升级完成后不会恢复原始槽。因此，当连接器重新启动并从复制槽请求最后已知偏移量时，PostgreSQL 无法返回信息。

您可以创建一个新的复制槽，但您需要做的不只是创建新槽来防止数据丢失。新复制槽只能提供您创建槽后发生的更改的 LSN；它无法提供升级前发生的事件的偏移量。当连接器重新启动时，它首先从 Kafka 偏移量主题请求最后已知偏移量。然后，它向复制槽发送请求，以返回从偏移量主题检索到的偏移量信息。但是，新复制槽无法提供连接器从预期位置恢复流式传输所需的信息。然后，连接器将跳过日志中的任何现有更改事件，仅从日志的最新位置恢复流式传输。这可能导致数据静默丢失：连接器不对跳过的事件发出记录，并且不提供任何信息来指示事件已被跳过。

有关如何执行 PostgreSQL 数据库升级以使 Debezium 能够继续捕获事件，同时最大程度地降低数据丢失风险的指导，请参阅以下过程。

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

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

请参阅 PostgreSQL 连接器属性的完整列表，这些属性可以在这些配置中指定。

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

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

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

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