Debezium系列之：使用Debezium接入PostgreSQL数据库数据到Kafka集群的详细技术文档

  • 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 驱动程序

相关链接：

连接器为捕获的每个行级插入、更新和删除操作生成一个更改事件，并为单独的 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 连接器配置属性的选项

选项描述
always连接器在启动时始终执行快照。快照完成后，连接器继续按上述顺序从第 3 步开始流式传输更改。此模式在以下情况下很有用：1.已知一些 WAL 段已被删除，不再可用。集群发生故障后，新的主节点已被提升。always快照模式确保连接器不会错过在新主节点提升之后但在新主节点上重新启动连接器之前所做的任何更改。
never连接器从不执行快照。以这种方式配置连接器时，其启动时的行为如下。如果 Kafka 偏移主题中存在先前存储的 LSN，则连接器会继续从该位置流式传输更改。如果没有存储 LSN，则连接器从在服务器上创建 PostgreSQL 逻辑复制槽的时间点开始流式传输更改。仅当您知道所有感兴趣的数据仍反映在 WAL 中时，never快照模式才有用。
initial_only连接器执行数据库快照并在流式传输任何更改事件记录之前停止。如果连接器已启动但在停止之前未完成快照，则连接器将重新启动快照进程并在快照完成时停止。
exported已弃用，所有模式都是无锁的。
custom自定义快照模式允许您注入自己的 io.debezium.connector.postgresql.spi.Snapshotter 接口实现。将 snapshot.custom.class 配置属性设置为 Kafka Connect 集群的类路径上的类，或者如果使用 EmbeddedEngine，则将其包含在 JAR 中。

3.Ad hoc snapshots

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

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

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

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

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

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

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

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

字段默认值值
typeincremental指定要运行的快照类型。
设置类型是可选的。目前，您只能请求增量快照。
data-collectionsN/A一个数组，其中包含要生成快照的表的完全限定名称。名称的格式与 signal.data.collection 配置选项的格式相同。

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 (_'<id>'_, _'<snapshotType>'_, '{"data-collections": ["_<tableName>_","_<tableName>_"],"type":"_<snapshotType>_"}');

例如，

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

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

下表描述了这些参数：

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

值描述
myschema.debezium_signal指定源数据库上的信令表的完全限定名称
ad-hoc-1id 参数指定一个任意字符串，该字符串被分配为信号请求的 id 标识符。使用此字符串将日志消息标识到信令表中的条目。 Debezium 不使用此字符串。相反，在快照期间，Debezium 生成自己的 id 字符串作为水印信号。
execute-snapshot指定类型参数指定信号要触发的操作。
data-collections信号数据字段的必需组件，它指定要包含在快照中的表名数组。该数组按表的完全限定名称列出表，使用的格式与您在 signal.data.collection 配置属性中指定连接器的信号表名称的格式相同。
incremental信号数据字段的可选类型组件，指定要运行的快照操作的种类。目前，唯一有效的选项是默认值增量。在您提交信令表的 SQL 查询中指定类型值是可选的。如果您未指定值，则连接器将运行增量快照。

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

{
    "before":null,
    "after": {
        "pk":"1",
        "value":"New data"
    },
    "source": {
        ...
        "snapshot":"incremental" 
    },
    "op":"r", 
    "ts_ms":"1620393591654",
    "transaction":null
}
ItemField name描述
1snapshot指定要运行的快照操作的类型。目前，唯一有效的选项是默认值incremental。在您提交给信令表的 SQL 查询中指定类型值是可选的。如果您未指定值，则连接器将运行增量快照。
2op指定事件类型。快照事件的值为 r，表示 READ 操作。

注意：

  • 用于 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

  • 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
    }
  ]
}

除非通过 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"
  }
}

三、数据更改事件

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

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

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

{
 "schema": { 
   ...
  },
 "payload": { 
   ...
 },
 "schema": { 
   ...
 },
 "payload": { 
   ...
 },
}

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

ItemField name描述
1schema第一个模式字段是事件键的一部分。它指定了一个 Kafka Connect 模式，该模式描述了事件键的有效负载部分中的内容。换句话说，第一个模式字段描述了主键的结构，如果表没有主键，则描述唯一键的结构，用于已更改的表。可以通过设置 message.key.columns 连接器配置属性来覆盖表的主键。在这种情况下，第一个模式字段描述了由该属性标识的键的结构。
2payload第一个有效负载字段是事件键的一部分。它具有前面架构字段所描述的结构，并且包含已更改行的键。
3schema第二个模式字段是事件值的一部分。它指定了描述事件值有效负载部分内容的 Kafka Connect 模式。换句话说，第二个模式描述了被改变的行的结构。通常，此模式包含嵌套模式。
4payload第二个有效负载字段是事件值的一部分。它具有前面架构字段所描述的结构，并且包含已更改行的实际数据。

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

从 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
    }
}

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

ItemField name描述
1schema值的架构，在连接器为特定表生成的每个更改事件中，更改事件的值架构都是相同的。
2name字段结构
3name源字段模式
4name数据库和表名称
5payload实际数据
6beforenull
7after新增数据值
8source源数据库相关信息：debezium版本，连接器类型，连接器名称，数据生成时间，是否是快照，数据库名称，数据库表，事务ID ，数据库偏移量
9opc = 创建，u = 更新，d = 删除，r = 读取（仅适用于快照），t = 截断，m = 消息
10ts_ms显示连接器处理事件的时间的可选字段。该时间基于运行 Kafka Connect 任务的 JVM 中的系统时钟。在源对象中，ts_ms 指示在数据库中进行更改的时间。通过将 payload.source.ts_ms 的值与 payload.ts_ms 的值进行比较，您可以确定源数据库更新和 Debezium 之间的延迟。

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)
    }
}

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

ItemField name描述
1before一个可选字段，包含数据库提交前行中的值。在此示例中，仅存在主键列 id，因为默认情况下，表的 REPLICA IDENTITY 设置为 DEFAULT。 + 要让更新事件包含行中所有列的先前值，您必须通过运行 ALTER TABLE customers REPLICA IDENTITY FULL 来更改客户表。
2after修改后数据值
3source源数据库相关信息：debezium版本，连接器类型，连接器名称，数据生成时间，是否是快照，数据库名称，数据库表，事务ID ，数据库偏移量
4opc = 创建，u = 更新，d = 删除，r = 读取（仅适用于快照），t = 截断，m = 消息
5ts_ms显示连接器处理事件的时间的可选字段。该时间基于运行 Kafka Connect 任务的 JVM 中的系统时钟。在源对象中，ts_ms 指示在数据库中进行更改的时间。通过将 payload.source.ts_ms 的值与 payload.ts_ms 的值进行比较，您可以确定源数据库更新和 Debezium 之间的延迟。

主键更新

更改行的主键字段的 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 
    }
}

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

ItemField name描述
1before可选字段，指定事件发生前行的状态。在删除事件值中，before 字段包含在使用数据库提交删除之前该行中的值。在此示例中，before 字段仅包含主键列，因为表的 REPLICA IDENTITY 设置为 DEFAULT。
2afternull
3source源数据库相关信息：debezium版本，连接器类型，连接器名称，数据生成时间，是否是快照，数据库名称，数据库表，事务ID ，数据库偏移量
4opc = 创建，u = 更新，d = 删除，r = 读取（仅适用于快照），t = 截断，m = 消息
5ts_ms显示连接器处理事件的时间的可选字段。该时间基于运行 Kafka Connect 任务的 JVM 中的系统时钟。在源对象中，ts_ms 指示在数据库中进行更改的时间。通过将 payload.source.ts_ms 的值与 payload.ts_ms 的值进行比较，您可以确定源数据库更新和 Debezium 之间的延迟。

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 
    }
}

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

ItemField name描述
1source源数据库相关信息：debezium版本，连接器类型，连接器名称，数据生成时间，是否是快照，数据库名称，数据库表，事务ID ，数据库偏移量
2opc = 创建，u = 更新，d = 删除，r = 读取（仅适用于快照），t = 截断，m = 消息
3ts_ms显示连接器处理事件的时间的可选字段。该时间基于运行 Kafka Connect 任务的 JVM 中的系统时钟。在源对象中，ts_ms 指示在数据库中进行更改的时间。通过将 payload.source.ts_ms 的值与 payload.ts_ms 的值进行比较，您可以确定源数据库更新和 Debezium 之间的延迟。

如果单个 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"
        }
    }
}

与其他事件类型不同，非事务性消息不会有任何关联的 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"
    }
}

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

ItemField name描述
1source源数据库相关信息：debezium版本，连接器类型，连接器名称，数据生成时间，是否是快照，数据库名称，数据库表，事务ID ，数据库偏移量
2opc = 创建，u = 更新，d = 删除，r = 读取（仅适用于快照），t = 截断，m = 消息
3ts_ms显示连接器处理事件的时间的可选字段。该时间基于运行 Kafka Connect 任务的 JVM 中的系统时钟。对于事务性消息事件，源对象的 ts_ms 属性指示在数据库中对事务性消息事件进行更改的时间。通过将 payload.source.ts_ms 的值与 payload.ts_ms 的值进行比较，您可以确定源数据库更新和 Debezium 之间的延迟。对于非事务性消息事件，源对象的 ts_ms 表示连接器遇到消息事件的时间，而 payload.ts_ms 表示连接器处理事件的时间。这种差异是由于 Postgres 的通用逻辑消息格式中不存在提交时间戳，并且非事务性逻辑消息之前没有 BEGIN 事件（具有时间戳信息）。
4message包含消息元数据的字段，前缀（文本）内容（基于二进制处理模式设置编码的字节数组）

五、数据类型映射

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

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

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

1.基本类型

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

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

PostgreSQL 数据类型文字类型（模式类型）语义类型（模式名称）和注释
BOOLEANBOOLEANn/a
BIT(1)BOOLEANn/a
BIT( > 1)BYTESio.debezium.data.Bits 长度模式参数包含一个表示位数的整数。生成的 byte[] 包含 little-endian 形式的位，并且大小调整为包含指定数量的位。例如，numBytes = n/8 + (n % 8 == 0 ? 0 : 1) 其中 n 是位数。
BIT VARYING[(M)]BYTESio.debezium.data.Bits 长度模式参数包含一个表示位数的整数（如果没有为列指定长度，则为 2^31 - 1）。结果 byte[] 包含 little-endian 形式的位，并根据内容调整大小。指定的大小（M）存储在 io.debezium.data.Bits 类型的长度参数中。
SMALLINT, SMALLSERIALINT16n/a
INTEGER, SERIALINT32n/a
BIGINT, BIGSERIAL, OIDINT64n/a
REALFLOAT32n/a
DOUBLE PRECISIONFLOAT64n/a
CHAR[(M)]STRINGn/a
VARCHAR[(M)]STRINGn/a
CHARACTER[(M)]STRINGn/a
CHARACTER VARYING[(M)]STRINGn/a
TIMESTAMPTZ, TIMESTAMP WITH TIME ZONESTRINGio.debezium.time.ZonedTimestamp 带有时区信息的时间戳的字符串表示形式，其中时区为 GMT。
TIMETZ, TIME WITH TIME ZONESTRINGio.debezium.time.ZonedTime 带有时区信息的时间值的字符串表示形式，其中时区为 GMT。
INTERVAL [P]INT64io.debezium.time.MicroDuration(default) 使用 365.25 / 12.0 公式计算每月平均天数的时间间隔的近似微秒数。
INTERVAL [P]STRINGio.debezium.time.Interval(when interval.handling.mode is set to string) 遵循模式 PYMDTHMS 的间隔值的字符串表示形式，例如 P1Y2M3DT4H5M6.78S。
BYTEABYTES or STRINGn/a 根据连接器的二进制处理模式设置，原始字节（默认）、base64 编码字符串或十六进制编码字符串。Debezium 仅支持 hex 值的 Postgres bytea_output 配置。请参阅有关 Postgres 二进制数据类型的文档。
JSON, JSONBSTRINGio.debezium.data.Json 包含 JSON 文档、数组或标量的字符串表示形式。
XMLSTRINGio.debezium.data.Xml 包含 XML 文档的字符串表示形式。
UUIDSTRINGio.debezium.data.Uuid 包含 PostgreSQL UUID 值的字符串表示形式。
POINTSTRUCTio.debezium.data.geometry.Point 包含具有两个 FLOAT64 字段 (x,y) 的结构。每个字段代表一个几何点的坐标。
LTREESTRINGio.debezium.data.Ltree 包含 PostgreSQL LTREE 值的字符串表示形式。
CITEXTSTRINGn/a
INETSTRINGn/a
INT4RANGESTRINGn/a 整数范围。
INT8RANGESTRINGn/a bigint 的范围。
NUMRANGESTRINGn/a 数值范围。
TSRANGESTRINGn/a 包含不带时区的时间戳范围的字符串表示形式。
TSTZRANGESTRINGn/a 包含具有本地系统时区的时间戳范围的字符串表示形式。
DATERANGESTRINGn/a 包含日期范围的字符串表示形式。它总是有一个独占的上限。
ENUMSTRINGio.debezium.data.Enum 包含 PostgreSQL ENUM 值的字符串表示。允许的值集在允许的模式参数中维护。

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 自适应时的映射

PostgreSQL 数据类型文字类型（模式类型）语义类型（模式名称）和注释
DATEINT32io.debezium.time.Date 表示自纪元以来的天数。
TIME(1), TIME(2), TIME(3)INT32io.debezium.time.Time 表示午夜过后的毫秒数，不包括时区信息。
TIME(4), TIME(5), TIME(6)INT64io.debezium.time.MicroTime 表示午夜过后的微秒数，不包括时区信息。
TIMESTAMP(1), TIMESTAMP(2), TIMESTAMP(3)INT64io.debezium.time.Timestamp 表示自纪元以来的毫秒数，不包括时区信息。
TIMESTAMP(4), TIMESTAMP(5), TIMESTAMP(6), TIMESTAMPINT64io.debezium.time.MicroTimestamp 表示自纪元以来的微秒数，不包括时区信息。

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

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

PostgreSQL 数据类型文字类型（模式类型）语义类型（模式名称）和注释
DATEINT32io.debezium.time.Date 表示自纪元以来的天数。
TIME([P])INT64io.debezium.time.MicroTime 以微秒为单位表示时间值，不包括时区信息。 PostgreSQL 允许精度 P 在 0-6 范围内以存储高达微秒的精度。
TIMESTAMP(1) , TIMESTAMP(2), TIMESTAMP(3)INT64io.debezium.time.Timestamp 表示经过纪元的毫秒数，不包括时区信息。
TIMESTAMP(4) , TIMESTAMP(5), TIMESTAMP(6), TIMESTAMPINT64io.debezium.time.MicroTimestamp 表示经过纪元的微秒数，不包括时区信息。

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

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

PostgreSQL 数据类型文字类型（模式类型）语义类型（模式名称）和注释
DATEINT32io.debezium.time.Date 表示自纪元以来的天数。
TIME([P])INT64org.apache.kafka.connect.data.Time 表示自午夜以来的毫秒数，不包括时区信息。 PostgreSQL 允许 P 在 0-6 范围内以存储高达微秒的精度，尽管当 P 大于 3 时这种模式会导致精度损失。
TIMESTAMP([P])INT64org.apache.kafka.connect.data.Timestamp 表示自纪元以来的毫秒数，不包括时区信息。 PostgreSQL 允许 P 在 0-6 范围内以存储高达微秒的精度，尽管当 P 大于 3 时这种模式会导致精度损失。

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 精确时的映射

PostgreSQL 数据类型文字类型（模式类型）语义类型（模式名称）和注释
NUMERIC[(M[,D])]BYTESorg.apache.kafka.connect.data.Decimal scale 模式参数包含一个整数，表示小数点移动了多少位。
DECIMAL[(M[,D])]BYTESorg.apache.kafka.connect.data.Decimal scale 模式参数包含一个整数，表示小数点移动了多少位。
MONEY[(M[,D])]BYTESorg.apache.kafka.connect.data.Decimal scale 模式参数包含一个整数，表示小数点移动了多少位。比例架构参数由 money.fraction.digits 连接器配置属性确定。

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

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

PostgreSQL 数据类型文字类型（模式类型）语义类型（模式名称）和注释
NUMERICSTRUCTio.debezium.data.VariableScaleDecimal 包含具有两个字段的结构：类型为 INT32 的标度包含传输值的标度，类型为 BYTES 的值包含未标度形式的原始值。
DECIMALSTRUCTio.debezium.data.VariableScaleDecimal 包含具有两个字段的结构：类型为 INT32 的标度包含传输值的标度，类型为 BYTES 的值包含未标度形式的原始值。

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

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

PostgreSQL 数据类型文字类型（模式类型）语义类型（模式名称）和注释
NUMERIC[(M[,D])]FLOAT64
DECIMAL[(M[,D])]FLOAT64
MONEY[(M[,D])]FLOAT64

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

PostgreSQL 数据类型文字类型（模式类型）语义类型（模式名称）和注释
NUMERIC[(M[,D])]STRING
DECIMAL[(M[,D])]STRING
MONEY[(M[,D])]STRING

当 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'

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

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

# REPLICATION
wal_level = logical
  • 指示服务器对预写日志使用逻辑解码。

需要在使用 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;

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

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

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

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

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

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

程序

  • 创建复制组。
CREATE ROLE <replication_group>;
  • 将表的原始所有者添加到组中。
GRANT REPLICATION_GROUP TO <original_owner>;
  • 将 Debezium 复制用户添加到该组。
GRANT REPLICATION_GROUP TO <replication_user>;
  • 将表的所有权转移到 <replication_group>。
ALTER TABLE <table_name> OWNER TO REPLICATION_GROUP;

为了让 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
  • 指示服务器允许在本地（即在服务器计算机上）复制 。
  • 指示服务器允许本地主机上的 <youruser> 使用 IPV4 接收复制更改。
  • 指示服务器允许本地主机上的 <youruser> 使用 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" 

  }
}
  • 向 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. 所需的连接器配置属性

nameNo default连接器的唯一名称。尝试使用相同名称再次注册将失败。所有 Kafka Connect 连接器都需要此属性。
connector.classNo default连接器的 Java 类的名称。始终为 PostgreSQL 连接器使用 io.debezium.connector.postgresql.PostgresConnector 值。
tasks.max1应为此连接器创建的最大任务数。 PostgreSQL 连接器始终使用单个任务，因此不使用此值，因此默认值始终是可以接受的。
plugin.namedecoderbufs安装在 PostgreSQL 服务器上的 PostgreSQL 逻辑解码插件的名称。支持的值是 decoderbufs 和 pgoutput。wal2json 插件已弃用并计划移除。
slot.namedebeziumPostgreSQL 逻辑解码槽的名称，它是为从特定数据库/模式的特定插件流式传输更改而创建的。服务器使用此插槽将事件流式传输到您正在配置的 Debezium 连接器。插槽名称必须符合 PostgreSQL 复制插槽命名规则，其中规定：“每个复制插槽都有一个名称，可以包含小写字母、数字和下划线字符。”
slot.drop.on.stopfalse当连接器以正常的、预期的方式停止时是否删除逻辑复制槽。默认行为是，当连接器停止时，复制槽仍然为连接器配置。当连接器重新启动时，具有相同的复制槽使连接器能够从中断处开始处理。仅在测试或开发环境中设置为 true。删除槽允许数据库丢弃 WAL 段。当连接器重新启动时，它会执行一个新的快照，或者它可以从 Kafka Connect 偏移量主题中的持久偏移量继续。
publication.namedbz_publication使用 pgoutput 时为流式更改创建的 PostgreSQL 发布的名称。如果此发布尚不存在并且包含所有表，则在启动时创建此发布。 Debezium 然后应用其自己的包含/排除列表过滤（如果已配置）来限制发布以更改感兴趣的特定表的事件。连接器用户必须具有超级用户权限才能创建此发布，因此通常最好在首次启动连接器之前创建发布。如果发布已经存在，无论是针对所有表还是配置了表的子集，Debezium 都会使用定义的发布。
database.hostnameNo defaultPostgreSQL 数据库服务器的 IP 地址或主机名。
database.port5432PostgreSQL 数据库服务器的整数端口号。
database.userNo default用于连接 PostgreSQL 数据库服务器的 PostgreSQL 数据库用户名。
database.passwordNo default连接到 PostgreSQL 数据库服务器时使用的密码。
database.dbnameNo default从中流式传输更改的 PostgreSQL 数据库的名称。
database.server.nameNo default为 Debezium 在其中捕获更改的特定 PostgreSQL 数据库服务器或集群标识并提供命名空间的逻辑名称。逻辑名称在所有其他连接器中应该是唯一的，因为它用作从该连接器接收记录的所有 Kafka 主题的主题名称前缀。数据库服务器逻辑名称中只能使用字母数字字符、连字符、点和下划线。不要更改此属性的值。如果您更改名称值，在重新启动后，连接器不会继续向原始主题发出事件，而是向名称基于新值的主题发出后续事件。
schema.include.listNo default一个可选的、以逗号分隔的正则表达式列表，与您要捕获更改的模式名称相匹配。未包含在 schema.include.list 中的任何模式名称都将被排除在捕获其更改之外。默认情况下，所有非系统模式都会捕获其更改。不要同时设置 schema.exclude.list 属性。
schema.exclude.listNo default一个可选的、以逗号分隔的正则表达式列表，与您不想捕获更改的模式名称匹配。任何名称未包含在 schema.exclude.list 中的模式都会捕获其更改，但系统模式除外。不要同时设置 schema.include.list 属性。
table.include.listNo default一个可选的、以逗号分隔的正则表达式列表，匹配您要捕获其更改的表的完全限定表标识符。未包含在 table.include.list 中的任何表都不会捕获其更改。每个标识符的格式为 schemaName.tableName。默认情况下，连接器会捕获每个模式中每个非系统表中的更改，这些表的更改正在被捕获。不要同时设置 table.exclude.list 属性。
table.exclude.listNo default一个可选的、以逗号分隔的正则表达式列表，匹配您不想捕获其更改的表的完全限定表标识符。未包含在 table.exclude.list 中的任何表都会捕获其更改。每个标识符的格式为 schemaName.tableName。不要同时设置 table.include.list 属性。
column.include.listNo default一个可选的、以逗号分隔的正则表达式列表，与应包含在更改事件记录值中的列的完全限定名称匹配。列的完全限定名称的格式为 schemaName.tableName.columnName。不要同时设置 column.exclude.list 属性。
column.exclude.listNo default一个可选的、以逗号分隔的正则表达式列表，与应从更改事件记录值中排除的列的完全限定名称匹配。列的完全限定名称的格式为 schemaName.tableName.columnName。不要同时设置 column.include.list 属性。
time.precision.modeadaptive时间、日期和时间戳可以用不同的精度表示：Adaptive 根据数据库列的类型，使用毫秒、微秒或纳秒精度值精确捕获数据库中的时间和时间戳值。Adaptive_time_microseconds 根据数据库列的类型使用毫秒、微秒或纳秒精度值精确地捕获数据库中的日期、日期时间和时间戳值。一个例外是 TIME 类型的字段，它总是以微秒为单位。connect 始终使用 Kafka Connect 内置的时间、日期和时间戳表示来表示时间和时间戳值，无论数据库列的精度如何，它们都使用毫秒精度。
decimal.handling.modeprecise指定连接器应如何处理 DECIMAL 和 NUMERIC 列的值：通过使用 java.math.BigDecimal 以二进制形式表示更改事件中的值来精确表示值。double 使用 double 值表示值，这可能会导致精度损失，但更易于使用。string 将值编码为格式化字符串，易于使用，但有关真实类型的语义信息会丢失。
hstore.handling.modemap指定连接器应如何处理 hstore 列的值：map 使用 MAP 表示值。json 使用 json 字符串表示值。此设置将值编码为格式化字符串，例如 {“key” : “val”}。
interval.handling.modenumeric指定连接器应如何处理间隔列的值：numeric 表示使用近似微秒数的间隔。string 使用字符串模式表示PYMDTHMS 精确表示间隔。例如：P1Y2M3DT4H5M6.78S。
database.sslmodedisable是否使用与 PostgreSQL 服务器的加密连接。选项包括：disable 使用未加密的连接。require 使用安全（加密）连接，如果无法建立连接，则会失败。verify-ca 的行为与 require 类似，但也会根据配置的证书颁发机构 (CA) 证书验证服务器 TLS 证书，如果没有找到有效的匹配 CA 证书，则会失败。verify-full 的行为类似于 verify-ca，但也验证服务器证书是否与连接器尝试连接的主机匹配。
database.sslcertNo default包含客户端 SSL 证书的文件的路径。
database.sslkeyNo default包含客户端 SSL 私钥的文件的路径。
database.sslpasswordNo default从 database.sslkey 指定的文件中访问客户端私钥的密码。
database.sslrootcertNo default包含验证服务器的根证书的文件的路径。
database.tcpKeepAlivetrue启用 TCP keep-alive 探测以验证数据库连接是否仍然有效。
tombstones.on.deletetrue控制删除事件后是否有墓碑事件。true - 删除操作由删除事件和后续的墓碑事件表示。false - 仅发出删除事件。删除源记录后，发出 tombstone 事件（默认行为）允许 Kafka 完全删除与已删除行的键相关的所有事件，以防主题启用日志压缩。
column.truncate.to.length.charsn/a一个可选的、以逗号分隔的正则表达式列表，匹配基于字符的列的完全限定名称。列的完全限定名称的格式为 schemaName.tableName.columnName。在更改事件记录中，如果这些列中的值长于属性名称中长度指定的字符数，则它们将被截断。您可以在单个配置中指定具有不同长度的多个属性。长度必须为正整数，例如 +column.truncate.to.20.chars。
column.mask.with.length.charsn/a一个可选的、以逗号分隔的正则表达式列表，匹配基于字符的列的完全限定名称。列的完全限定名称的格式为 schemaName.tableName.columnName。在更改事件值中，指定表列中的值将替换为星号 (*) 字符的长度数。您可以在单个配置中指定具有不同长度的多个属性。长度必须是正整数或零。当您指定零时，连接器会将值替换为空字符串。
column.mask.hash.hashAlgorithm.with.salt.salt; column.mask.hash.v2.hashAlgorithm.with.salt.saltn/a一个可选的、以逗号分隔的正则表达式列表，匹配基于字符的列的完全限定名称。列的完全限定名称的格式为 <schemaName>..。在生成的更改事件记录中，指定列的值将替换为假名。假名由应用指定的 hashAlgorithm 和 salt 产生的散列值组成。基于所使用的散列函数，参照完整性得以保持，而列值被替换为假名。 Java Cryptography Architecture Standard Algorithm Name Documentation 的 MessageDigest 部分描述了支持的散列函数。在以下示例中，CzQMA0cB5K 是随机选择的盐。column.mask.hash.SHA-256.with.salt.CzQMA0cB5K = inventory.orders.customerName, inventory.shipment.customerName如有必要，假名会自动缩短为列的长度。连接器配置可以包括指定不同哈希算法和盐的多个属性。根据使用的 hashAlgorithm、选择的 salt 和实际数据集，生成的数据集可能不会被完全屏蔽。如果值在不同的地方或系统中进行散列，则应使用散列策略版本 2 来确保保真度。
publication.autocreate.modeall_tables仅当使用 pgoutput 插件进行流式更改时才适用。该设置决定了如何创建发布。可能的设置是：all_tables - 如果发布存在，连接器将使用它。如果发布不存在，则连接器会为数据库中连接器正在捕获更改的所有表创建发布。这要求具有执行复制权限的数据库用户也具有创建发布的权限。这是通过 CREATE PUBLICATION <publication_name> FOR ALL TABLES; 授予的。disabled - 连接器不尝试创建发布。数据库管理员或配置为执行复制的用户必须在运行连接器之前创建发布。如果连接器找不到发布，连接器将引发异常并停止。filtered - 如果发布存在，连接器将使用它。如果不存在发布，则连接器会为与由 database.exclude.list、schema.include.list、schema.exclude.list 和 table.include.list 连接器配置属性指定的当前过滤器配置匹配的表创建新发布.例如：CREATE PUBLICATION <publication_name> FOR TABLE <tbl1, tbl2, tbl3>。
binary.handling.modebytes指定二进制 (bytea) 列应如何在更改事件中表示：bytes 将二进制数据表示为字节数组。base64 将二进制数据表示为 base64 编码的字符串。hex 将二进制数据表示为十六进制编码 (base16) 字符串。
schema.name.adjustment.modeavro指定应如何调整架构名称以与连接器使用的消息转换器兼容。可能的设置：avro 将不能在 Avro 类型名称中使用的字符替换为下划线。none 不应用任何调整。
truncate.handling.modeskip指定是否应该传播 TRUNCATE 事件（仅在使用 pgoutput 插件和 Postgres 11 或更高版本时可用）：skip 导致这些事件被忽略（默认）。include 导致包含这些事件。此选项已弃用。使用 skipped.operations 代替它。

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

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

属性默认值描述
convertersNo default枚举连接器可以使用的自定义转换器实例的符号名称的逗号分隔列表。例如，isbn您必须设置转换器属性以使连接器能够使用自定义转换器。对于您为连接器配置的每个转换器，您还必须添加一个 .type 属性，该属性指定实现转换器接口的类的完全限定名称。 .type 属性使用以下格式：<converterSymbolicName>.type 例如，isbn.type：io.debezium.test.IsbnConverter如果要进一步控制已配置转换器的行为，可以添加一个或多个配置参数以将值传递给转换器。要将任何其他配置参数与转换器关联，请在参数名称前加上转换器的符号名称。例如，isbn.schema.name: io.debezium.postgresql.type.Isbn
snapshot.modeinitial指定连接器启动时执行快照的条件：initial - 仅当没有为逻辑服务器名称记录偏移时，连接器才会执行快照。always - 每次连接器启动时，连接器都会执行快照。never - 连接器从不执行快照。以这种方式配置连接器时，其启动时的行为如下。如果 Kafka 偏移主题中存在先前存储的 LSN，则连接器会继续从该位置流式传输更改。如果没有存储 LSN，则连接器从在服务器上创建 PostgreSQL 逻辑复制槽的时间点开始流式传输更改。仅当您知道所有感兴趣的数据仍反映在 WAL 中时，从不快照模式才有用。initial_only - 连接器执行初始快照，然后停止，不处理任何后续更改。exported - 已弃用custom - 连接器根据 snapshot.custom.class 属性的设置执行快照，该属性是 io.debezium.connector.postgresql.spi.Snapshotter 接口的自定义实现。
snapshot.include.collection.listAll tables specified in table.include.list一个可选的、以逗号分隔的正则表达式列表，与要包含在快照中的表的完全限定名称 (<schemaName>.<tableName>) 匹配。指定的项目必须在连接器的 table.include.list 属性中命名。仅当连接器的 snapshot.mode 属性设置为 never 以外的值时，此属性才会生效。此属性不影响增量快照的行为。
event.processing.failure.handling.modefail指定连接器在处理事件期间应如何对异常做出反应：fail 传播异常，指示有问题的事件的偏移量，并导致连接器停止。warn 记录有问题的事件的偏移量，跳过该事件并继续处理。skip 跳过有问题的事件并继续处理。
max.batch.size10240指定连接器处理的每批事件的最大大小的正整数值。
max.queue.size20240正整数值，指定阻塞队列可以保存的最大记录数。当 Debezium 读取从数据库流式传输的事件时，它会将事件放入阻塞队列中，然后再将它们写入 Kafka。在连接器接收消息的速度快于将消息写入 Kafka 的速度或 Kafka 不可用时，阻塞队列可以为从数据库读取更改事件提供背压。当连接器定期记录偏移量时，将忽略队列中保存的事件。始终将 max.queue.size 的值设置为大于 max.batch.size 的值。
max.queue.size.in.bytes0一个长整数值，指定阻塞队列的最大容量（以字节为单位）。默认情况下，没有为阻塞队列指定容量限制。要指定队列可以使用的字节数，请将此属性设置为正长值。如果还设置了 max.queue.size，则当队列的大小达到任一属性指定的限制时，将阻止写入队列。例如，如果设置 max.queue.size=1000，max.queue.size.in.bytes=5000，则在队列包含 1000 条记录后，或队列中的记录量大后，写入队列被阻塞达到 5000 字节。
poll.interval.ms1000正整数值，指定连接器在开始处理一批事件之前应等待新更改事件出现的毫秒数。默认为 1000 毫秒或 1 秒。
status.update.interval.ms10000向服务器发送复制连接状态更新的频率，以毫秒为单位。该属性还控制在数据库关闭的情况下检查数据库状态以检测死连接的频率。
heartbeat.interval.ms0控制连接器向 Kafka 主题发送心跳消息的频率。默认行为是连接器不发送心跳消息。心跳消息对于监视连接器是否从数据库接收更改事件很有用。心跳消息可能有助于减少连接器重新启动时需要重新发送的更改事件的数量。要发送心跳消息，请将此属性设置为正整数，表示心跳消息之间的毫秒数。当正在跟踪的数据库中有许多更新但只有极少数更新与连接器正在捕获更改的表和模式相关时，需要心跳消息。在这种情况下，连接器照常从数据库事务日志中读取，但很少将更改记录发送到 Kafka。这意味着没有向 Kafka 提交偏移更新，并且连接器没有机会将最新检索到的 LSN 发送到数据库。数据库保留包含已由连接器处理的事件的 WAL 文件。发送心跳消息使连接器能够将最新检索到的 LSN 发送到数据库，这允许数据库回收不再需要的 WAL 文件使用的磁盘空间。
schema.refresh.modecolumns_diff指定触发刷新表的内存模式的条件。columns_diff 是最安全的模式。它确保内存中的模式始终与数据库表的模式保持同步。columns_diff_exclude_unchanged_toast 指示连接器在与从传入消息派生的架构存在差异时刷新内存中架构缓存，除非未更改的 TOASTable 数据完全解释了差异。如果经常更新的表具有 TOASTed 数据，而这些数据很少是更新的一部分，则此设置可以显着提高连接器性能。但是，如果从表中删除 TOASTable 列，则内存中的模式可能会过时。
snapshot.delay.msNo default连接器启动时执行快照之前连接器应等待的时间间隔（以毫秒为单位）。如果您在集群中启动多个连接器，此属性对于避免快照中断很有用，这可能会导致连接器重新平衡。
snapshot.fetch.size10240在快照期间，连接器分批读取表内容。此属性指定批处理中的最大行数。
slot.stream.paramsNo default分号分隔的参数列表传递给配置的逻辑解码插件。例如，add-tables=public.table,public.table2;include-lsn=true。
slot.max.retries6如果连接到复制槽失败，这是连续尝试连接的最大次数。
slot.retry.delay.ms10000 (10 seconds)当连接器无法连接到复制槽时，重试尝试之间等待的毫秒数。
retriable.restart.connector.wait.ms10000 (10 seconds)发生可重试错误后重新启动连接器之前等待的毫秒数。
skipped.operationst在流式传输期间将跳过的操作类型的逗号分隔列表。操作包括：c 表示插入/创建，u 表示更新，d 表示删除，t 表示截断，none 表示不跳过任何操作。默认情况下，会跳过截断操作。
signal.data.collectionNo default value用于向连接器发送信号的数据集合的完全限定名称。使用以下格式指定集合名称:<schemaName>.
incremental.snapshot.chunk.size1024连接器在增量快照块期间获取并读入内存的最大行数。增加块大小提供了更高的效率，因为快照运行的快照查询更少，但更大的大小。然而，更大的块大小也需要更多的内存来缓冲快照数据。将块大小调整为在您的环境中提供最佳性能的值。
xmin.fetch.interval.ms0从复制槽读取 XMIN 的频率（以毫秒为单位）。 XMIN 值提供了新复制槽可以从哪里开始的下限。默认值 0 禁用跟踪 XMIN 跟踪。

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

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

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

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

1.Snapshot metrics

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

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

属性类型描述
LastEventstring连接器读取的最后一个快照事件。
MilliSecondsSinceLastEventlong自连接器读取并处理最新事件以来的毫秒数。
TotalNumberOfEventsSeenlong自上次启动或重置以来，此连接器已看到的事件总数。
NumberOfEventsFilteredlong已被连接器上配置的包含/排除列表过滤规则过滤的事件数。
CapturedTablesstring[]连接器捕获的表列表。
QueueTotalCapacityint用于在快照程序和主 Kafka Connect 循环之间传递事件的队列长度。
QueueRemainingCapacityint用于在快照程序和主 Kafka Connect 循环之间传递事件的队列的可用容量。
TotalTableCountint快照中包含的表总数。
RemainingTableCountint快照尚未复制的表数。
SnapshotRunningboolean快照是否已启动。
SnapshotAbortedboolean快照是否已中止。
SnapshotCompletedboolean快照是否完成。
SnapshotDurationInSecondslong到目前为止，快照已拍摄的总秒数，即使未完成也是如此。
RowsScannedMap<String, Long>包含为快照中的每个表扫描的行数的映射。在处理过程中，表格会逐渐添加到地图中。每扫描 10,000 行并在完成表时更新。
MaxQueueSizeInByteslong队列的最大缓冲区（以字节为单位）。max.queue.size.in.bytes如果设置为正长值，则此指标可用。
CurrentQueueSizeInByteslong队列中的当前记录量（以字节为单位）。

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

属性类型描述
ChunkIdstring当前快照块的标识符。
ChunkFromstring定义当前块的主键集的下限。
ChunkTostring定义当前块的主键集的上限。
TableFromstring当前快照表的主键集的下限。
TableTostring当前快照表的主键集的上限。

2.Streaming metrics

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

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

属性类型描述
LastEventstring连接器读取的最后一个流事件。
MilliSecondsSinceLastEventlong自连接器读取并处理最新事件以来的毫秒数。
TotalNumberOfEventsSeenlong自上次启动或指标重置以来，此连接器已看到的事件总数。
TotalNumberOfCreateEventsSeenlong自上次启动或指标重置以来，此连接器已看到的创建事件总数。
TotalNumberOfUpdateEventsSeenlong自上次启动或指标重置以来，此连接器已看到的更新事件总数。
TotalNumberOfDeleteEventsSeenlong自上次启动或指标重置以来，此连接器已看到的删除事件总数。
NumberOfEventsFilteredlong已被连接器上配置的包含/排除列表过滤规则过滤的事件数。
CapturedTablesstring[]连接器捕获的表列表。
QueueTotalCapacityint用于在流媒体和主 Kafka Connect 循环之间传递事件的队列长度。
QueueRemainingCapacityint队列的空闲容量，用于在流媒体和 Kafka Connect 主循环之间传递事件。
Connectedboolean表示连接器当前是否连接到数据库服务器的标志。
MilliSecondsBehindSourcelong上次更改事件的时间戳与处理它的连接器之间的毫秒数。这些值将包含运行数据库服务器和连接器的机器上的时钟之间的任何差异。
NumberOfCommittedTransactionslong已提交的已处理事务数。
SourceEventPositionMap<String, String>最后接收到的事件的坐标。
LastTransactionIdstring最后处理的事务的事务标识符。
MaxQueueSizeInByteslong队列的最大缓冲区（以字节为单位）。max.queue.size.in.bytes如果设置为正长值，则此指标可用。
CurrentQueueSizeInByteslong队列中的当前记录量（以字节为单位）。

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

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

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

Logo
Kafka开源项目指南

Kafka开源项目指南提供详尽教程，助开发者掌握其架构、配置和使用，实现高效数据流管理和实时处理。它高性能、可扩展，适合日志收集和实时数据处理，通过持久化保障数据安全，是企业大数据生态系统的核心。

更多推荐

Kafka入门（一） 概述、部署与API的简单使用

Kafka概述、部署与API的简单使用

avatar Kafka开源项目指南

基于canal和kafka同步，实现binlog同步ElasticSearch

文章目录前言elasticsearch 安装canal安装canal-adapter 安装及配置mysql 安装zk及kafaka安装查看效果注意事项前言中间件版本elasticsearch7.5.2canal1.1.4client-adapter1.1.5-alpha-1zookeeper3.4.13kafka2.6.0mysql5.7.31elasticsearch 安装{"settings"

avatar Kafka开源项目指南

基于 Iceberg 的湖仓一体架构在 B 站的实践

背景在B站，每天都有PB级的数据注入到大数据平台，经过离线或实时的ETL建模后，提供给下游的分析、推荐及预测等场景使用。面对如此大规模的数据，如何高效低成本地满足下游数据的分析需求，一直是我们重点的工作方向。我们之前的数据处理流程基本上是这样的：采集端将客户端埋点、服务端埋点、日志、业务数据库等数据收集到HDFS、Kafka等存储系统中，然后通过Hive、Spark、Fl...

avatar Kafka开源项目指南