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)）提交的更改。

每当服务器提交事务时，一个单独的服务器进程会调用逻辑解码插件的回调函数。该函数处理事务的更改，将它们转换为特定格式（对于 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 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 能够回收存储空间，同时确保主题包含完整的数据集并可用于重新加载基于键的状态。

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

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

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

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

消息事件表示一个通用逻辑解码消息已直接插入 WAL，通常使用 pg_logical_emit_message 函数。消息键是具有单个字段 prefix 的 Struct，包含插入消息时指定的名称。对于事务性消息，消息值如下所示：

与其他事件类型不同，非事务性消息将不具有任何关联的 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 支持微秒精度，当数据库列的小数秒精度值大于 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 来映射时间戳、日期时间（datetime）和时间值。

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

TIMESTAMP 类型表示不带时区信息的时间戳。此类列根据 UTC 转换为等效的 Kafka Connect 值。例如，当 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，它同时包含 transferred value 的值和精度。

当 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 遵循以下规则来处理 TOAST 值：

如果在数据库模式中为列指定了默认值，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（日志序列号）的形式维护当前位置，该 LSN 是指向 Debezium 连接器正在使用的 WAL 中某个位置的指针。这有助于 PostgreSQL 在 Debezium 处理之前保留 WAL。单个复制槽只能由单个使用者或进程存在，因为不同的使用者可能有不同的状态，并且可能需要来自不同位置的数据。

由于一个复制槽只能由一个连接器使用，因此为每个 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 连接器的配置示例，该连接器连接到端口 5432、IP 地址 192.168.99.100 上的 PostgreSQL 服务器，其逻辑名称为 fulfillment。通常，您通过设置连接器可用的配置属性来配置 Debezium PostgreSQL 连接器，将其存储在 JSON 文件中。

您可以选择为数据库中的一部分模式和表生成事件。此外，您还可以选择忽略、屏蔽或截断包含敏感数据、尺寸大于指定大小或您不需要的列。

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

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

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

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

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