Debezium JDBC 连接器

默认情况下，Debezium 源连接器会生成复杂、分层的更改事件。当 Debezium 连接器与其他 JDBC 接收器连接器实现一起使用时，您可能需要应用 ExtractNewRecordState 单个消息转换 (SMT) 来展平事件的有效负载，以便接收器实现能够消费它们。如果您运行 Debezium JDBC 接收器连接器，则无需部署 SMT，因为 Debezium 接收器连接器可以直接消费原生的 Debezium 更改事件，而无需进行转换。

当 JDBC 接收器连接器从 Debezium 源连接器消费复杂更改事件时，它会提取原始 insert 或 update 事件的 after 部分的值。当接收器连接器消费删除事件时，不会考虑事件的有效负载的任何部分。

Debezium JDBC 接收器连接器保证它从 Kafka 主题消费的事件至少被处理一次。

您可以跨多个 Kafka Connect 任务运行 Debezium JDBC 接收器连接器。要跨多个任务运行连接器，请将 tasks.max 配置属性设置为您希望连接器使用的任务数量。Kafka Connect 运行时会启动指定数量的任务，并为每个任务运行一个连接器实例。多个任务可以通过并行读取和处理多个源主题的更改来提高性能。

为了使 Debezium JDBC 接收器连接器能够正确地将入站消息字段中的数据类型映射到出站消息字段中的数据类型，连接器需要有关源事件中存在的每个字段的数据类型的信息。该连接器支持跨不同数据库方言的广泛的列类型映射。为了正确地从事件字段中的 type 元数据转换目标列类型，连接器应用为源数据库定义的数据类型映射。您可以通过在源连接器配置中设置 column.propagate.source.type 或 datatype.propagate.source.type 选项来增强连接器解析列数据类型的方式。启用这些选项后，Debezium 会包含额外的参数元数据，这有助于 JDBC 接收器连接器更准确地解析目标列的数据类型。

要使 Debezium JDBC 接收器连接器能够处理 Kafka 主题中的事件，Kafka 主题消息键（如果存在）必须是原始数据类型或 Struct。此外，源消息的有效负载必须是 Struct，它具有一个展平的结构，没有嵌套的 struct 类型，或者具有符合 Debezium 的复杂、分层结构的嵌套 struct 布局。

如果 Kafka 主题中事件的结构不符合这些规则，您必须实现自定义的单个消息转换，将源事件的结构转换为可用的格式。

默认情况下，Debezium JDBC 接收器连接器不会将源事件中的任何字段转换为事件的主键。不幸的是，稳定的主键缺失可能会使事件处理复杂化，具体取决于您的业务需求，或者当接收器连接器使用 upsert 语义时。要定义一致的主键，您可以将连接器配置为使用下表中描述的主键模式之一。

当消费到 DELETE 或 tombstone 事件时，Debezium JDBC 接收器连接器可以删除目标数据库中的行。默认情况下，JDBC 接收器连接器不启用删除模式。

如果您希望连接器删除行，则必须在连接器配置中显式设置 delete.enabled=true。要使用此模式，您还必须将 primary.key.fields 设置为 none 以外的值。上述配置是必需的，因为删除是基于主键映射执行的，所以如果目标表没有主键映射，连接器就无法删除行。

Debezium JDBC 接收器连接器可以执行幂等写入，从而使其能够反复重放相同的记录，而不会改变最终的数据库状态。

要使连接器能够执行幂等写入，您必须在连接器中显式将 insert.mode 设置为 upsert。upsert 操作根据指定的主键是否已存在，执行为 update 或 insert。

如果主键值已存在，操作将更新行中的值。如果指定的主键值不存在，insert 将添加新行。

每个数据库方言处理幂等写入的方式不同，因为没有 SQL 标准来处理 upsert 操作。下表显示了 Debezium 支持的数据库方言的 upsert DML 语法。

您可以使用以下模式演进与 Debezium JDBC 接收器连接器：

当 schema.evolution 设置为 basic 时，连接器会根据传入事件的结构自动创建或修改目标数据库表。

当第一次从主题接收事件，并且目标表尚不存在时，Debezium JDBC 接收器连接器会使用事件的键或记录的结构来解析表的列结构。如果启用了模式演进，连接器会在应用 DML 事件到目标表之前，准备并执行 CREATE TABLE SQL 语句。

当 Debezium JDBC 连接器从主题接收事件时，如果记录的模式结构与目标表的模式结构不同，连接器会使用事件的键或其模式结构来识别哪些列是新的，需要添加到数据库表中。如果启用了模式演进，连接器会在应用 DML 事件到目标表之前，准备并执行 ALTER TABLE SQL 语句。由于更改列数据类型、删除列和调整主键可能被视为危险操作，因此连接器不允许执行这些操作。

每个字段的模式决定了列是 NULL 还是 NOT NULL。模式还定义了每列的默认值。如果连接器尝试创建一个具有您不想要的 nullability 设置或默认值的表，您必须手动提前创建表，或者在接收器连接器处理事件之前调整相关字段的模式。要调整 nullability 设置或默认值，您可以引入自定义的单个消息转换，在管道中应用更改，或修改源数据库中定义的列状态。

字段的数据类型是根据预定义的映射集解析的。有关更多信息，请参阅 JDBC 字段类型。

Debezium JDBC 接收器连接器通过构建 DDL（模式更改）或 DML（数据更改）SQL 语句来消费 Kafka 消息，这些语句会在目标数据库上执行。默认情况下，连接器使用源主题和事件字段的名称作为目标表中表和列名称的基础。构造的 SQL 不会自动用引号分隔标识符以保留原始字符串的大小写。因此，默认情况下，目标数据库中表或列名称的文本大小写完全取决于数据库在未指定大小写时如何处理名称字符串。

例如，如果目标数据库方言是 Oracle，并且事件主题是 orders，则目标表将创建为 ORDERS，因为 Oracle 在未引用名称时默认使用大写名称。同样，如果目标数据库方言是 PostgreSQL，并且事件主题是 ORDERS，则目标表将创建为 orders，因为 PostgreSQL 在未引用名称时默认使用小写名称。

要显式保留 Kafka 事件中的表和字段名称的大小写，请在连接器配置中将 quote.identifiers 属性的值设置为 true。当此选项设置为 true 时，如果传入事件的主题名为 orders，并且目标数据库方言是 Oracle，则连接器将创建一个名为 orders 的表，因为构造的 SQL 将表名定义为 "orders"。启用引号会导致连接器创建列名时产生相同的行为。

Debezium 的 JDBC 接收器连接器利用连接池来提高性能。连接池旨在建立一组初始连接，维护指定数量的连接，并根据需要有效地将连接分配给应用程序。但是，当连接在池中长时间空闲时会出现问题，如果它们保持不活动状态超过数据库配置的空闲超时阈值，可能会导致超时。

为了减轻空闲连接线程触发超时的可能性，连接池提供了一种机制，该机制会定期验证每个连接的活动状态。此验证可确保连接保持活动状态，并防止数据库将其标记为空闲。在发生网络中断的情况下，如果 Debezium 尝试使用已终止的连接，连接器会提示池生成新连接。

默认情况下，Debezium JDBC 接收器连接器不执行空闲超时测试。但是，您可以通过设置 hibernate.c3p0.idle_test_period 属性来配置连接器，使其以指定的时间间隔执行超时测试。例如：

Debezium JDBC 接收器连接器使用 Hibernate C3P0 连接池。您可以通过在 hibernate.c3p0.* 配置命名空间中设置属性来定制 CP30 连接池。在前面的示例中，设置 hibernate.c3p0.idle_test_period 属性会将连接池配置为每 300 秒执行一次空闲超时测试。应用配置后，连接池将开始每五分钟评估一次未使用的连接。

Debezium JDBC 接收器连接器支持将逻辑向量数据类型从源事件直接映射到接收器目标，前提是目标数据库支持可比较的表示。

如果源中的向量字段具有逻辑名称，Debezium 会使用它来确定适当的映射。对于某些数据库，连接器会在 io.debezium.data.vector.* 命名空间下识别以下逻辑名称：

当连接器处理消息时，如果存在这些逻辑类型，连接器会检查目标数据库中是否存在相应的映射。如果存在映射，则应用它；否则，它将根据 Kafka Connect 模式类型将数据序列化为字符串。

例如，配置为将数据发送到 PostgreSQL 数据库接收器的连接器会将具有 io.debezium.data.FloatVector 逻辑名称的字段映射到 halfvector 列类型，使用特殊的覆盖映射。相比之下，如果接收器数据库中不存在直接映射，例如 Oracle 接收器，则连接器将默认将向量数据序列化为其原始字符串格式。

并非所有支持的数据库版本都提供对特殊向量类型的支持。接收器数据库必须满足以下要求才能支持逻辑向量类型的直接映射：

早期版本的 MariaDB 或 MySQL，以及没有 pgvector 的 PostgreSQL 不支持特殊向量类型。

除了前面表格中显示的原始和逻辑映射之外，如果更改事件的源是 Debezium 源连接器，那么列类型以及其长度、精度和小数位的解析还可以通过启用列或数据类型传播来进一步影响。列和数据类型传播有助于确定传入数据的结构和数据类型如何被翻译并应用于接收器目标。如前文 数据和列类型映射 中所述，您可以通过在源连接器配置中设置以下属性之一来强制传播：

Debezium JDBC 接收器连接器仅应用具有更高优先级的那些值。

为了说明传播如何影响连接器映射数据类型，让我们看一个例子。以下示例显示了可能包含在更改事件中的字段模式。

由于发出事件的源连接器配置为使用列或数据类型传播，因此事件包含了指定列 type 和 length 的参数。

如果源连接器未启用传播，则 type 和 length 参数将缺失，Debezium JDBC 接收器连接器将默认将 type 字段中的 INT8 值映射到 Types.SMALLINT 列类型。根据目标数据库的 SQL 方言，连接器随后会将 JDBC Types.SMALLINT 类型解析为多个逻辑类型之一。例如，对于 MySQL 接收器目标，JDBC 连接器默认将 Types.SMALLINT 转换为 TINYINT 列类型，不指定长度。

但是，由于源连接器中启用了类型传播，它发出的事件包含了 type 和 length，提供了更具体的映射说明。因此，Debezium JDBC 接收器连接器不使用默认映射，而是使用给定的参数值来优化映射，并在接收器数据库中创建一个类型为 TINYINT(1) 的列。

Debezium JDBC 连接器提供了两种命名转换，您可以将其应用于更改事件中的主题或字段名称的命名样式。可选地，您还可以使用命名转换为转换后的名称添加前缀或后缀。

您可以将命名转换配置为使用以下命名样式之一：

连接器提供以下命名转换：

通常，您通过提交一个 JSON 请求来注册 Debezium JDBC 连接器，该请求指定连接器的配置属性。以下示例显示了一个用于注册 Debezium JDBC 接收器连接器实例的 JSON 请求，该实例从名为 orders 的主题消费事件，并使用最常见的配置设置。

JDBC 连接器配置设置说明

有关您可以为 Debezium JDBC 连接器设置的完整配置属性列表，请参阅 JDBC 连接器属性。

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

Kafka Connect 支持直通配置，使您能够通过将某些属性直接从连接器配置传递来修改底层系统的行为。默认情况下，一些 Hibernate 属性通过 JDBC 连接器的 连接属性（例如，connection.url、connection.username 和 connection.pool.*_size）以及通过连接器的 运行时属性（例如，use.time.zone、quote.identifiers）公开。

如果您想自定义其他 Hibernate 行为，可以通过在连接器配置中添加使用 hibernate.* 命名空间的属性来利用直通机制。例如，为了帮助 Hibernate 解析目标数据库的类型和版本，您可以添加 hibernate.dialect 属性并将其设置为数据库的完全限定类名，例如 org.hibernate.dialect.MariaDBDialect。