Debezium系列之：Debezium2.X之Debezium JDBC连接器

靳富

2023-12-01

一、概述

Debezium JDBC 连接器是一个 Kafka Connect 接收器连接器实现，它可以使用来自多个源主题的事件，然后使用 JDBC 驱动程序将这些事件写入关系数据库。
此连接器支持多种数据库方言，包括 Db2、MySQL、Oracle、PostgreSQL 和 SQL Server。

二、JDBC连接器的工作原理

Debezium JDBC 连接器是 Kafka Connect 接收器连接器，因此需要 Kafka Connect 运行时。连接器定期轮询它订阅的 Kafka 主题，使用来自这些主题的事件，然后将事件写入配置的关系数据库。连接器通过使用 upsert 语义和基本模式演变来支持幂等写入操作。

支持以下功能：

使用复杂的 Debezium 更改事件
至少一次交付
多项任务
数据和列类型映射
主键处理
删除模式
幂等写入
模式演变
引用和区分大小写

1.使用复杂的Debezium更改事件

默认情况下，Debezium 源连接器会产生复杂的分层更改事件。
当 Debezium 连接器与其他 JDBC 接收器连接器实现一起使用时，您可能需要应用ExtractNewRecordState单一消息转换 (SMT) 来展平更改事件的有效负载，以便接收器实现可以使用它们。
如果您运行 Debezium JDBC 接收器连接器，则无需部署 SMT，因为 Debezium 接收器连接器可以直接使用本机 Debezium 更改事件，而无需使用转换。
after当 JDBC 接收器连接器使用来自 Debezium 源连接器的复杂更改事件时，它会从原始insert或update事件部分提取值。当接收器连接器使用删除事件时，不会查询事件有效负载的任何部分。

2.至少一次交付

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

3.多项任务

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

4.数据和列类型映射

为了使 Debezium JDBC 接收器连接器能够将数据类型从入站消息字段正确映射到出站消息字段，连接器需要有关源事件中存在的每个字段的数据类型的信息。该连接器支持跨不同数据库方言的各种列类型映射。为了从事件字段中的元数据正确转换目标列类型type，连接器应用为源数据库定义的数据类型映射。
column.propagate.source.type您可以通过设置或来增强连接器解析列数据类型的方式datatype.propagate.source.type源连接器配置中的选项。当您启用这些选项时，Debezium 包含额外的参数元数据，这有助于 JDBC 接收器连接器更准确地解析目标列的数据类型。
对于 Debezium JDBC 接收器连接器来处理来自 Kafka 主题的事件，Kafka 主题消息键（如果存在）必须是原始数据类型或Struct. 此外，源消息的有效负载必须是Struct具有没有嵌套struct类型的扁平结构，或者struct符合 Debezium 复杂层次结构的嵌套布局。
如果 Kafka 主题中的事件结构不符合这些规则，您必须实施自定义单消息转换以将源事件的结构转换为可用格式。

5.主键处理

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

模式	描述
none	创建表时没有指定主键字段。
kafka	主键由以下三列组成：__connect_topic，__connect_partition，__connect_offset这些列的值来自 Kafka 事件的坐标。
record_key	主键由 Kafka 事件的键组成。如果主键是原始类型，则通过设置属性指定要使用的列的名称primary.key.fields。如果主键是结构类型，则结构中的字段映射为主键的列。您可以使用该primary.key.fields属性将主键限制为列的子集。
record_value	主键由 Kafka 事件的值组成。因为 Kafka 事件的值始终是 a Struct，默认情况下，值中的所有字段都成为主键的列。要使用主键中的字段子集，请设置该primary.key.fields属性以在要从中派生主键列的值中指定以逗号分隔的字段列表。

6.删除模式

Debezium JDBC 接收器连接器可以在使用DELETEor逻辑删除事件时删除目标数据库中的行。默认情况下，JDBC 接收器连接器不启用删除模式。
如果要支持删除行，请通过在连接器配置中设置delete.enabled=true明确启用它。但是，要使用此模式，您必须设置primary.key.fields为以外的值none。上述配置是必须的，因为删除是根据主键映射执行的，所以如果目标表没有主键映射，连接器将无法删除行。

7.幂等写入

Debezium JDBC 接收器连接器支持幂等写入，允许重复重放相同的记录并使最终数据库状态保持一致。为了支持幂等写入，JDBC 接收器连接器必须显式配置insert.mode为upsert. 根据指定的主键是否已存在，操作upsert被应用为update或 insert。如果主键值已经存在，则该操作会更新行中的值。如果指定的主键值不存在，则insert添加一个新行。

每种数据库方言以不同方式处理幂等写入，因为没有用于更新插入操作的 SQL 标准。下面说明了支持的数据库方言使用的 upsert 数据库特定的 DML 语法：

数据库	更新语法
DB2	MERGE …
MySQL	INSERT … ON DUPLICATE KEY UPDATE …
Oracle	MERGE …
PostgreSQL	INSERT … ON CONFLICT … DO UPDATE SET …
SQL Server	MERGE …

8.模式演变

Debezium JDBC 接收器连接器支持以下模式演变模式：

模式	描述
none	连接器不执行任何 DDL 架构演变。
basic	连接器会自动检测事件负载中存在但目标表中不存在的字段。连接器更改目标表以添加新字段。

当schema.evolution设置为basic时，连接器会根据传入事件的结构自动创建或更改目标数据库表。
当第一次从主题接收到事件并且目标表尚不存在时，Debezium JDBC 接收器连接器使用事件的键或记录的模式结构来解析表的列结构。如果启用模式演变，连接器会CREATE TABLE在将 DML 事件应用于目标表之前准备并执行 SQL 语句。
每个字段的模式决定了一个列是NULL还是NOT NULL。该架构还为每一列定义了默认值。如果连接器尝试创建一个带有可空性设置或不需要的默认值的表，您必须提前手动创建该表，或者在接收器连接器处理事件之前调整关联字段的架构。要调整可空性设置或默认值，您可以引入一个自定义的单一消息转换，在管道中应用更改，或修改源数据库中定义的列状态。
字段的数据类型是根据一组预定义的映射来解析的。

注意：

当您将新字段引入目标数据库中已存在的表的事件结构时，您必须将新字段定义为可选字段，或者这些字段必须具有在数据库架构中指定的默认值。如果要从目标表中删除字段，请使用以下选项之一：
手动删除字段。
为字段分配默认值。
将字段定义为可为空。

9.引用和区分大小写

Debezium JDBC 接收器连接器通过构建在目标数据库上执行的 DDL（架构更改）或 DML（数据更改）SQL 语句来使用 Kafka 消息。默认情况下，连接器使用源主题和事件字段的名称作为目标表中表名和列名的基础。构造的 SQL 不会自动用引号分隔标识符以保留原始字符串的大小写。因此，默认情况下，目标数据库中表名或列名的文本大小写完全取决于数据库在未指定大小写时如何处理名称字符串。
例如，如果目标数据库方言是 Oracle 并且事件的主题是orders，则目标表将被创建为ORDERS因为当名称未被引用时 Oracle 默认为大写名称。类似地，如果目标数据库方言是 PostgreSQL 并且事件的主题是ORDERS，则目标表将被创建为orders因为当名称未被引用时 PostgreSQL 默认为小写名称。
通过在连接器配置中设置quote.identifiers为true，您可以显式设置表和字段名称的大小写以保留 Kafka 事件中存在的大小写。因此，如果传入事件是针对名为的主题orders并且目标数据库方言是 Oracle，如果启用了引用，也就是说如果quote.identifiers设置为true，连接器将创建一个名为的表orders，因为构造的 SQL 定义了表的名称作为"orders"。启用引用时，创建列名的行为以相同的方式工作。

三、数据类型映射

Debezium JDBC 接收器连接器通过使用逻辑或原始类型映射系统解析列的数据类型。原始类型包括整数、浮点数、布尔值、字符串和字节等值。通常，这些类型仅用特定的 Kafka Connect 类型代码表示Schema。逻辑数据类型通常是复杂类型，包括值，例如Struct具有一组固定字段名称和模式的基于的类型，或使用特定编码表示的值，例如自纪元以来的天数。

以下示例显示了原始数据类型和逻辑数据类型的代表性结构：

原始字段模式

{
  "schema": {
    "type": "INT64"
  }
}

逻辑字段模式

[
  "schema": {
    "type": "INT64",
    "name": "org.apache.kafka.connect.data.Date"
  }
]

Kafka Connect 不是这些复杂的逻辑类型的唯一来源。事实上，Debezium 源连接器生成的更改事件具有具有相似逻辑类型的字段，以表示各种不同的数据类型，包括但不限于时间戳、日期，甚至 JSON 数据。

Debezium JDBC 接收器连接器使用这些原始和逻辑类型将列的类型解析为表示列类型的 JDBC SQL 代码。然后，底层 Hibernate 持久性框架使用这些 JDBC SQL 代码将列的类型解析为所用方言的逻辑数据类型。下表说明了 Kafka Connect 和 JDBC SQL 类型之间以及 Debezium 和 JDBC SQL 类型之间的原始和逻辑映射。实际的最终列类型因每种数据库类型而异。

Kafka Connect原始类型和列数据类型之间的映射
Kafka Connect 逻辑类型和列数据类型之间的映射
Debezium 逻辑类型和列数据类型之间的映射
Debezium特定的逻辑类型和列数据类型之间的映射

Kafka Connect原始类型和列数据类型之间的映射：

原始类型	JDBC SQL类型
INT8	Types.TINYINT
INT16	Types.SMALLINT
INT32	Types.INTEGER
INT64	Types.BIGINT
FLOAT32	Types.FLOAT
FLOAT64	Types.DOUBLE
BOOLEAN	Types.BOOLEAN
STRING	Types.CHAR, Types.NCHAR, Types.VARCHAR, Types.NVARCHAR
BYTES	Types.VARBINARY

Kafka Connect 逻辑类型和列数据类型之间的映射：

逻辑类型	JDBC SQL类型
org.apache.kafka.connect.data.Decimal	Types.DECIMAL
org.apache.kafka.connect.data.Date	Types.DATE
org.apache.kafka.connect.data.Time	Types.TIMESTAMP
org.apache.kafka.connect.data.Timestamp	Types.TIMESTAMP

Debezium 逻辑类型和列数据类型之间的映射：

逻辑类型	JDBC SQL类型
io.debezium.time.Date	Types.DATE
io.debezium.time.Time	Types.TIMESTAMP
io.debezium.time.MicroTime	Types.TIMESTAMP
io.debezium.time.NanoTime	Types.TIMESTAMP
io.debezium.time.ZonedTime	Types.TIME_WITH_TIMEZONE
io.debezium.time.Timestamp	Types.TIMESTAMP
io.debezium.time.MicroTimestamp	Types.TIMESTAMP
io.debezium.time.NanoTimestamp	Types.TIMESTAMP
io.debezium.time.ZonedTimestamp	Types.TIMESTAMP_WITH_TIMEZONE
io.debezium.data.VariableScaleDecimal	Types.DOUBLE

如果数据库不支持带时区的时间或时间戳，则映射解析为不带时区的等效映射。

Debezium特定的逻辑类型和列数据类型之间的映射

逻辑类型	MySQL SQL 类型	PostgreSQL SQL 类型	SQL Server SQL 类型
io.debezium.data.Bits	bit(n)	bit(n) or bit varying	varbinary(n)
io.debezium.data.Enum	enum	Types.VARCHAR	n/a
io.debezium.data.Json	json	json	n/a
io.debezium.data.EnumSet	set	n/a	n/a
io.debezium.time.Year	year(n)	n/a	n/a
io.debezium.time.MicroDuration	n/a	interval	n/a
io.debezium.data.Ltree	n/a	ltree	n/a
io.debezium.data.Uuid	n/a	uuid	n/a
io.debezium.data.Xml	n/a	xml	xml

除了上面的原始和逻辑映射之外，如果更改事件的源是 Debezium 源连接器，则列类型的分辨率及其长度、精度和比例可能会受到启用列或数据类型的进一步影响传播。要强制传播，必须在源连接器配置中设置以下属性之一：

column.propagate.source.type
datatype.propagate.source.type

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

例如，假设更改事件中包含以下字段架构：

启用列或数据类型传播的 Debezium 更改事件字段模式

{
  "schema": {
    "type": "INT8",
    "parameters": {
      "__debezium.source.column.type": "TINYINT",
      "__debezium.source.column.length": "1"
    }
  }
}

在前面的示例中，如果未设置架构参数，则 Debezium JDBC 接收器连接器会将此字段映射到列类型Types.SMALLINT。Types.SMALLINT可以有不同的逻辑数据库类型，这取决于数据库。对于 MySQL，该示例将转换为TINYINT没有指定长度的列类型。如果为源连接器启用了列或数据类型传播，则 Debezium JDBC 接收器连接器使用映射信息来优化数据类型映射过程并创建类型为TINYINT(1)的列。

通常，当源数据库和接收器数据库使用相同类型的数据库时，使用列或数据类型传播的效果会大得多。我们一直在寻找改进这种跨异构数据库映射的方法，当前的类型系统允许我们根据反馈继续改进这些映射。

四、部署

要部署 Debezium JDBC 连接器，您需要安装 Debezium JDBC 连接器存档、配置连接器，然后通过将其配置添加到 Kafka Connect 来启动连接器。

先决条件：

安装了Apache ZooKeeper、Apache Kafka和Kafka Connect 。
目标数据库已安装并配置为接受 JDBC 连接。

程序：

下载 Debezium JDBC 连接器插件JDBC连接器。
将文件提取到您的 Kafka Connect 环境中。
（可选）从 Maven Central 下载 JDBC 驱动程序并将下载的驱动程序文件解压缩到包含 JDBC 接收器连接器 JAR 文件的目录。注意：Oracle 和 Db2 的驱动程序不会自动随 JDBC 接收器连接器一起提供，必须手动安装。
将驱动程序 JAR 文件添加到已安装 JDBC 接收器连接器的路径中。
确保安装 JDBC 接收器连接器的路径是Kafka Connect 的plugin.path。
重新启动 Kafka Connect 进程以获取新的 JAR 文件。

五、Debezium JDBC 连接器配置

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

Debezium JDBC连接器配置

{
    "name": "jdbc-connector",  
    "config": {
        "connector.class": "io.debezium.connector.jdbc.JdbcSinkConnector",  
        "tasks.max": "1",  
        "connection.url": "jdbc:postgresql://localhost/db",  
        "connection.username": "pguser",  
        "connection.password": "pgpassword",  
        "insert.mode": "upsert",  
        "delete.enabled": "true",  
        "primary.key.mode": "record_key",  
        "schema.evolution": "basic",  
        "database.time_zone": "UTC"  
    }
}

向 Kafka Connect 服务注册时分配给连接器的名称。
JDBC 接收器连接器类的名称。
为此连接器创建的最大任务数。
连接器用于连接到它写入的接收器数据库的 JDBC URL。
用于身份验证的数据库用户的名称。
用于身份验证的数据库用户的密码。
连接器使用的插入模式。
允许删除数据库中的记录。
指定用于解析主键列的方法。
使连接器能够发展目标数据库的架构。
指定编写时态字段类型时使用的时区。

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

连接到数据库。
从订阅的 Kafka 主题中消费事件。
将事件写入配置的数据库。

六、连接器属性

Debezium JDBC 接收器连接器有几个配置属性，可以使用它们来实现满足您需要的连接器行为。许多属性都有默认值。有关属性的信息组织如下：

连接属性：

属性	默认值	描述
connection.url	无默认	用于连接到数据库的 JDBC 连接 URL。
connection.username	无默认	连接器用于连接到数据库的数据库用户帐户的名称。
connection.password	无默认	连接器用于连接到数据库的密码。
connection.pool.min_size	5	指定池中的最小连接数。
connection.pool.min_size	32	指定池维护的最大并发连接数。
connection.pool.acquire_increment	32	指定在连接池超过其最大大小时连接器尝试获取的连接数。
connection.pool.timeout	1800	指定未使用的连接在被丢弃之前保留的秒数。

运行时属性：

属性	默认值	描述
database.time_zone	UTC	指定插入 JDBC 时间值时使用的时区。
delete.enabled	false	指定连接器是否处理DELETE逻辑删除事件并从数据库中删除相应的行。使用此选项需要您将设置primary.key.mode为record.key。
insert.mode	insert	指定用于将事件插入数据库的策略。以下选项可用：insert指定所有事件都应构造INSERT基于 SQL 语句。仅当不使用主键时，或者当您可以确定不会对具有现有主键值的行进行更新时，才使用此选项。update指定所有事件都应构造UPDATE基于 SQL 语句。仅当您可以确定连接器仅接收适用于现有行的事件时才使用此选项。upsert指定连接器使用upsert语义将事件添加到表中。即如果主键不存在，则连接器执行操作INSERT，如果键存在，则连接器执行操作UPDATE。当需要幂等写入时，应将连接器配置为使用此选项。
primary.key.mode	none	指定连接器如何从事件中解析主键列。none指定不创建主键列。kafka指定连接器使用 Kafka 坐标作为主键列。关键坐标是根据事件的主题名称、分区和偏移量定义的，并映射到具有以下名称的列：__connect_topic、__connect_partition、__connect_offset。record_key指定主键列源自事件的记录键。如果记录键是原始类型，则primary.key.fields需要该属性指定主键列的名称。如果记录键是结构类型，primary.key.fields则该属性是可选的，可用于将事件键中的列子集指定为表的主键。record_value指定主键列源自事件的值。可以设置该primary.key.fields属性以将主键定义为事件值的字段子集；否则默认使用所有字段。
primary.key.fields	无默认	主键列的名称或从中派生主键的以逗号分隔的字段列表。当primary.key.mode设置为record_key并且事件的键是原始类型时，预计此属性指定要用于键的列名。当使用非原始键primary.key.mode设置为或时，预计此属性指定键或值中以逗号分隔的字段名称列表。如果使用非原始键设置为或，并且未指定此属性，则连接器根据指定的模式从记录键或记录值的所有字段派生主键。
quote.identifiers	false	指定生成的 SQL 语句是否使用引号来分隔表名和列名。
schema.evolution	none	指定连接器如何发展目标表架构。none指定连接器不发展目标模式。basic指定发生基本进化。连接器通过将传入事件的记录模式与数据库表结构进行比较，将缺失的列添加到表中。
table.name.format	${topic}	指定一个字符串，该字符串根据事件的主题名称确定目标表名称的格式。占位符${topic}替换为主题名称。

可扩展属性：

属性	默认值	描述
column.naming.strategy	i.d.c.j.n.DefaultColumnNamingStrategy	指定 ColumnNamingStrategy 实现的完全限定类名，连接器使用该实现从事件字段名中解析列名。默认情况下，连接器使用字段名称作为列名称。
table.naming.strategy	i.d.c.j.n.DefaultTableNamingStrategy	TableNamingStrategy指定连接器用于从传入事件主题名称解析表名称的实现的完全限定类名称。默认行为是：${topic}将配置属性中的占位符替换table.name.format为事件的主题。通过将点 (.) 替换为下划线 (_) 来清理表名。

七、常见问题

1.是否需要ExtractNewRecordState单条消息转换？

不，这实际上是 Debezium JDBC 连接器与其他实现的区别因素之一。虽然连接器能够像其竞争对手一样摄取扁平化事件，但它也可以原生摄取 Debezium 的复杂更改事件结构，而不需要任何特定类型的转换。

2.如果更改了列的类型，或者重命名或删除了列，这是否由模式演化处理？

不，Debezium JDBC 连接器不会对现有列进行任何更改。连接器支持的模式演变非常基础。它只是将事件结构中的字段与表的列列表进行比较，然后添加任何尚未定义为表中的列的字段。如果列的类型或默认值更改，连接器不会在目标数据库中调整它们。如果一列被重命名，旧列保持原样，连接器将一个具有新名称的列附加到表中；但是，旧列中包含数据的现有行保持不变。这些类型的架构更改应手动处理。

3.如果列的类型未解析为我想要的类型，我如何强制映射到不同的数据类型？

Debezium JDBC 连接器使用复杂的类型系统来解析列的数据类型。有关此类型系统如何将特定字段的架构定义解析为 JDBC 类型的详细信息，请参阅数据和列类型映射部分。如果要应用不同的数据类型映射，请手动定义表以显式获取首选列类型。

4.如何在不更改 Kafka 主题名称的情况下为表名指定前缀或后缀？

为了向目标表名称添加前缀或后缀，请调整table.name.format连接器配置属性以应用所需的前缀或后缀。例如，要为所有表名添加前缀jdbc_，请将table.name.format配置属性指定为值jdbc_${topic}。如果连接器订阅了名为的主题orders，则生成的表将创建为jdbc_orders。

5.为什么有些列会自动引用，即使未启用标识符引用？

在某些情况下，可能会显式引用特定的列或表名称，即使quote.identifiers未启用也是如此。当列名或表名以特定约定开头或使用否则将被视为非法语法的特定约定时，这通常是必需的。例如，当primary.key.mode设置为kafka时，如果列名被引用，一些数据库只允许列名以下划线开头。引用行为是特定于方言的，并且在不同类型的数据库之间有所不同。

八、实战案例

Debezium系列之：基于Debezium JDBC connector消费Topic数据到Mysql数据库

更多内容请参考博主Debezium专栏：

Debezium专栏：Debezium系列文章汇总