网站域名多少钱一年,凡客诚品售后服务有哪些,设计理念,个人如何做商城网站Debezium日常分享系列之#xff1a;Debezium3版本Debezium connector for JDBC 概述JDBC连接器的工作原理消费复杂的Debezium变更事件至少一次的传递多个任务数据和列类型映射主键处理删除模式幂等写入模式演化引用和大小写敏感性连接空闲超时数据类型映射部署Debezium JDBC连… Debezium日常分享系列之Debezium3版本Debezium connector for JDBC 概述JDBC连接器的工作原理消费复杂的Debezium变更事件至少一次的传递多个任务数据和列类型映射主键处理删除模式幂等写入模式演化引用和大小写敏感性连接空闲超时数据类型映射部署Debezium JDBC连接器配置JDBC连接器hibernate.*透传属性常见问题更多Debezium技术内容 概述
Debezium JDBC连接器是一个Kafka Connect接收器连接器实现可以从多个源主题消费事件然后通过使用JDBC驱动程序将这些事件写入关系型数据库。该连接器支持各种数据库方言包括Db2、MySQL、Oracle、PostgreSQL和SQL Server。
JDBC连接器的工作原理
Debezium JDBC连接器是一个Kafka Connect接收器连接器因此需要Kafka Connect运行时。该连接器定期轮询其订阅的Kafka主题从这些主题中消费事件然后将事件写入配置的关系型数据库。该连接器通过使用upsert语义和基本模式演化来支持幂等写操作。
Debezium JDBC连接器提供以下功能
支持从多个源主题消费事件并写入关系型数据库。支持各种数据库方言包括Db2、MySQL、Oracle、PostgreSQL和SQL Server。支持幂等写操作通过使用upsert语义和基本模式演化。可以定期轮询Kafka主题并消费事件。提供基于JDBC驱动程序的可靠数据传输和写入。可以配置数据的格式和映射规则。支持监视和跟踪变更事件。可以处理大规模和高吞吐量的数据。具有可扩展性和容错性可以处理故障和重启情况。提供详细的日志记录和监控功能。
消费复杂的Debezium变更事件
默认情况下Debezium源连接器生成复杂的分层变更事件。当将Debezium连接器与其他JDBC接收器连接器实现一起使用时您可能需要应用ExtractNewRecordState单个消息转换SMT来展平变更事件的有效载荷以便可以由接收器实现消费。如果运行Debezium JDBC接收器连接器则无需部署SMT因为Debezium接收器连接器可以直接消费原生的Debezium变更事件无需使用转换。
当JDBC接收器连接器从Debezium源连接器消费复杂变更事件时它从原始插入或更新事件的after部分提取值。当接收器消费删除事件时不会查看事件的有效负载的任何部分。
重要 Debezium JDBC接收器连接器未设计用于从模式更改主题读取。如果您的源连接器配置为捕获模式更改请在JDBC连接器配置中设置topics或topics.regex属性以便连接器不从模式更改主题消费。
至少一次的传递
Debezium JDBC接收器连接器保证从Kafka主题消费的事件至少被处理一次。
多个任务
您可以在多个Kafka Connect任务上运行Debezium JDBC接收器连接器。要在多个任务上运行连接器请将tasks.max配置属性设置为您希望连接器使用的任务数量。Kafka Connect运行时会启动指定数量的任务并在每个任务上运行一个连接器实例。多个任务可以通过并行读取和处理来自多个源主题的更改从而提高性能。
数据和列类型映射
为了使Debezium JDBC接收器连接器能够正确地将入站消息字段的数据类型映射到出站消息字段连接器需要有关源事件中每个字段的数据类型的信息。该连接器支持不同数据库方言之间的广泛的列类型映射。为了正确地将事件字段中的类型元数据转换为目标列类型连接器应用了为源数据库定义的数据类型映射。您可以通过在源连接器配置中设置column.propagate.source.type或datatype.propagate.source.type选项来改进连接器解析列的数据类型的方式。启用这些选项时Debezium会包含额外的参数元数据帮助JDBC接收器连接器更准确地解析目标列的数据类型。
为了使Debezium JDBC接收器连接器能够处理来自Kafka主题的事件Kafka主题消息键如果存在必须是原始数据类型或Struct。此外源消息的有效负载必须是一个具有没有嵌套结构类型的扁平结构或者符合Debezium复杂的分层结构的嵌套结构布局。
如果Kafka主题中的事件结构不符合这些规则您必须实现自定义的单个消息转换将源事件的结构转换为可用的格式。
主键处理
默认情况下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 事件的值始终是结构体因此默认情况下值中的所有字段都将成为主键的列。要使用主键中的字段子集请设置 primary.key.fields 属性以指定要从中派生主键列的值中的字段的逗号分隔列表。record_header主键由 Kafka 事件的标头组成。Kafka 事件的标头可能包含多个标头每个标头可以是结构或原始数据类型连接器会将这些标头组成一个结构。因此此结构中的所有字段都将成为主键的列。要使用主键中的字段子集请设置 primary.key.fields 属性以指定要从中派生主键列的值中的字段的逗号分隔列表。
重要 某些数据库方言可能会在将primary.key.mode设置为kafka并将schema.evolution设置为basic时抛出异常。当方言将STRING数据类型映射到可变长度字符串数据类型如TEXT或CLOB并且方言不允许主键列具有无限长度时会出现此异常。为了避免这个问题请在您的环境中应用以下设置
不要将schema.evolution设置为basic。提前创建数据库表和主键映射。
如果某个列映射到目标数据库中不允许作为主键的数据类型那么在primary.key.fields中需要明确列出这些列。
删除模式
当消费DELETE或tombstone事件时Debezium JDBC接收器连接器可以在目标数据库中删除行。默认情况下JDBC接收器连接器不启用删除模式。 如果您想让连接器删除行必须在连接器配置中显式设置delete.enabledtrue。为了使用此模式您还必须将primary.key.fields设置为非none的值。之前的配置是必要的因为删除操作是基于主键映射执行的所以如果目标表没有主键映射连接器将无法删除行。
幂等写入
Debezium JDBC接收器连接器可以执行幂等写入使其能够重复播放相同的记录而不改变最终的数据库状态。
要使连接器执行幂等写入必须显式设置连接器的insert.mode为upsert。upsert操作将根据指定的主键是否已存在应用为更新或插入操作。
如果主键值已经存在则操作将更新行中的值。如果指定的主键值不存在则插入操作将添加新行。
每个数据库方言对幂等写入的处理方式不同因为没有针对upsert操作的SQL标准。下表显示了Debezium支持的数据库方言的upsert DML语法
DialectUpsert 语法Db2MERGE …MySQLINSERT … ON DUPLICATE KEY UPDATE …OracleMERGE …PostgreSQLINSERT … ON CONFLICT … DO UPDATE SET …SQL ServerMERGE …
模式演化
您可以将以下模式演化模式与 Debezium JDBC 接收器连接器结合使用
模式描述none连接器不执行任何 DDL 模式演变。basic连接器会自动检测事件负载中存在但目标表中不存在的字段。连接器会修改目标表以添加新字段。
当将schema.evolution设置为basic时连接器根据传入事件的结构自动创建或更改目标数据库表。
当首次从主题接收到事件并且目标表尚不存在时Debezium JDBC接收器连接器使用事件的键或记录的模式结构来解析表的列结构。如果启用了模式演化连接器会在将DML事件应用于目标表之前准备并执行CREATE TABLE SQL语句。
当Debezium JDBC连接器从主题接收到事件时如果记录的模式结构与目标表的模式结构不同连接器将使用事件的键或其模式结构来识别哪些列是新添加的并必须添加到数据库表中。如果启用了模式演化连接器会在将DML事件应用于目标表之前准备并执行ALTER TABLE SQL语句。由于更改列数据类型、删除列和调整主键可能被认为是危险操作连接器禁止执行这些操作。
每个字段的模式确定列是否为NULL或NOT NULL。模式还定义了每个列的默认值。如果连接器尝试创建具有不希望的nullability设置或默认值的表则必须提前手动创建表或在接收器连接器处理事件之前调整关联字段的模式。要调整nullability设置或默认值可以引入自定义的单个消息转换在管道中应用更改或修改源数据库中定义的列状态。
字段的数据类型是根据预定义的映射集解析的。
重要 当您向已存在于目标数据库中的表的事件结构中引入新字段时必须将新字段定义为可选的或者这些字段必须在数据库模式中指定默认值。如果您想从目标表中删除一个字段可以使用以下选项之一
手动删除该字段。删除该列。为该字段分配默认值。将该字段定义为可空。
引用和大小写敏感性
Debezium JDBC接收器连接器通过构建在目标数据库上执行的DDL模式更改或DML数据更改SQL语句来消费Kafka消息。默认情况下连接器使用源主题和事件字段的名称作为目标表和列名称的基础。构建的SQL语句不会自动使用引号引用标识符以保留原始字符串的大小写。因此默认情况下目标数据库中表或列名称的大小写取决于数据库在未指定大小写时如何处理名称字符串。
例如如果目标数据库方言是Oracle事件的主题是orders那么目标表将被创建为ORDERS因为Oracle在未使用引号引用时默认使用大写名称。类似地如果目标数据库方言是PostgreSQL事件的主题是ORDERS那么目标表将被创建为orders因为PostgreSQL在未使用引号引用时默认使用小写名称。
为了显式保留Kafka事件中存在的表和字段名称的大小写在连接器配置中将quote.identifiers属性的值设置为true。当设置了此选项时当传入的事件属于名为orders的主题并且目标数据库方言是Oracle时连接器将创建一个名为orders的表因为构建的SQL将表名定义为orders。启用引号引用时连接器创建列名时会产生相同的行为。
连接空闲超时
Debezium的JDBC接收器连接器利用连接池来提高性能。连接池被设计用于建立初始一组连接保持指定数量的连接并根据需要有效地为应用程序分配连接。然而当连接在连接池中闲置时可能会出现一个挑战如果它们在数据库配置的空闲超时阈值之后仍处于非活动状态可能会触发超时。
为了减轻空闲连接线程触发超时的潜在问题连接池提供了定期验证每个连接活动性的机制。这个验证确保连接保持活动并防止数据库将其标记为闲置。在网络中断的情况下如果Debezium尝试使用已终止的连接连接器会提示连接池生成一个新的连接。
默认情况下Debezium的JDBC接收器连接器不进行空闲超时测试。然而您可以通过设置hibernate.c3p0.idle_test_period属性来配置连接器以指定的间隔请求连接池执行超时测试。例如
{hibernate.c3p0.idle_test_period: 300
}Debezium的JDBC接收器连接器使用Hibernate C3P0连接池。您可以通过设置hibernate.c3p0.*配置名称空间中的属性来自定义CP30连接池。在前面的示例中设置hibernate.c3p0.idle_test_period属性将配置连接池每300秒执行一次空闲超时测试。在应用配置后连接池将开始每五分钟评估未使用的连接。
数据类型映射
Debezium的JDBC接收器连接器通过使用逻辑或基本类型映射系统来解析列的数据类型。基本类型包括整数、浮点数、布尔值、字符串和字节等值。通常这些类型仅使用特定的Kafka Connect Schema类型代码表示。逻辑数据类型更常见地是复杂类型包括具有固定字段名称和模式的基于结构的类型或者使用特定编码表示的值例如自纪元以来的天数。
以下示例显示了基本和逻辑数据类型的代表结构
基本字段模式
{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 方言特定逻辑类型与列数据类型之间的映射
表 1. Kafka Connect 原语和列数据类型之间的映射
Primitive TypeJDBC SQL TypeINT8Types.TINYINTINT16Types.SMALLINTINT32Types.INTEGERINT64Types.BIGINTFLOAT32Types.FLOATFLOAT64Types.DOUBLEBOOLEANTypes.BOOLEANSTRINGTypes.CHAR, Types.NCHAR, Types.VARCHAR, Types.NVARCHARBYTESTypes.VARBINARY
表 2. Kafka Connect 逻辑类型和列数据类型之间的映射
Primitive TypeJDBC SQL Typeorg.apache.kafka.connect.data.DecimalTypes.DECIMALorg.apache.kafka.connect.data.DateTypes.DATEorg.apache.kafka.connect.data.TimeTypes.TIMESTAMPorg.apache.kafka.connect.data.TimestampTypes.TIMESTAMP
表 3. Debezium 逻辑类型和列数据类型之间的映射
Primitive TypeJDBC SQL Typeio.debezium.time.DateTypes.DATEio.debezium.time.TimeTypes.TIMESTAMPio.debezium.time.MicroTimeTypes.TIMESTAMPio.debezium.time.NanoTimeTypes.TIMESTAMPio.debezium.time.ZonedTimeTypes.TIME_WITH_TIMEZONEio.debezium.time.TimestampTypes.TIMESTAMPio.debezium.time.MicroTimestampTypes.TIMESTAMPio.debezium.time.NanoTimestampTypes.TIMESTAMPio.debezium.time.ZonedTimestampTypes.TIMESTAMP_WITH_TIMEZONEio.debezium.data.VariableScaleDecimalTypes.DOUBLE
重要的 如果数据库不支持时间或带有时区的时间戳则映射将解析为没有时区的等效时间戳。
表 4. Debezium 方言特定逻辑类型与列数据类型之间的映射 除了上面的原始和逻辑映射外如果变更事件的源是Debezium源连接器则可以通过启用列或数据类型传播进一步影响列类型的分辨率以及其长度、精度和比例。要强制传播必须在源连接器配置中设置以下属性之一
column.propagate.source.typedatatype.propagate.source.type
Debezium JDBC sink连接器应用具有更高优先级的值。例如假设变更事件中包含以下字段架构
Debezium 启用列或数据类型传播来更改事件字段架构
{schema: {type: INT8,parameters: {__debezium.source.column.type: TINYINT,__debezium.source.column.length: 1}}
}在上述示例中如果没有设置模式参数Debezium JDBC sink连接器将此字段映射为Types.SMALLINT列类型。Types.SMALLINT可以根据数据库方言的不同而具有不同的逻辑数据库类型。对于MySQL在示例中列类型转换为未指定长度的TINYINT列类型。如果为源连接器启用了列或数据类型传播则Debezium JDBC sink连接器使用映射信息来细化数据类型映射过程并创建一个类型为TINYINT(1)的列。
注意 通常情况下当源数据库和目标数据库使用相同类型的数据库时使用列或数据类型传播的效果会更显著。我们一直在努力改进跨异构数据库的映射方式当前的类型系统允许我们根据反馈不断完善这些映射。
部署
要部署Debezium JDBC连接器您需要安装Debezium JDBC连接器归档文件配置连接器并通过将其配置添加到Kafka Connect来启动连接器。
先决条件
已安装和配置Kafka Connect、Kafka和Zookeeper。已安装并配置了一个目标数据库以接受JDBC连接。
步骤
下载Debezium。将文件提取到您的Kafka Connect环境中。可选地从Maven Central下载JDBC驱动程序并将下载的驱动程序文件提取到包含JDBC sink连接器JAR文件的目录中。 注意Oracle和Db2的驱动程序未包含在JDBC sink连接器中。您必须手动下载驱动程序并安装它们。 将驱动程序JAR文件添加到已安装JDBC sink连接器的路径中。确保您安装JDBC sink连接器的路径是CLASSPATH的一部分。重新启动Kafka Connect进程以获取新的JAR文件。
Debezium JDBC连接器配置
通常情况下您可以通过提交一个JSON请求来注册Debezium JDBC连接器该请求指定了连接器的配置属性。以下示例显示了一个JSON请求用于注册一个Debezium JDBC sink连接器实例该实例从名为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, topics: orders }
}JDBC 连接器配置设置描述
使用 Kafka Connect 服务注册连接器时分配给连接器的名称。JDBC 接收器连接器类的名称。为此连接器创建的最大任务数。连接器用于连接到其写入的接收器数据库的 JDBC URL。用于身份验证的数据库用户的名称。用于身份验证的数据库用户的密码。连接器使用的 insert.mode。启用数据库中记录的删除。有关更多信息请参阅 delete.enabled 配置属性。指定用于解析主键列的方法。有关更多信息请参阅 primary.key.mode 配置属性。使连接器能够演变目标数据库的架构。有关更多信息请参阅 schema.evolution 配置属性。指定写入时间字段类型时使用的时区。要使用的主题列表以逗号分隔。
您可以通过POST命令将此配置发送到正在运行的Kafka Connect服务。服务记录配置并启动一个或多个sink连接器任务执行以下操作
连接到数据库。从订阅的Kafka主题消费事件。将事件写入配置的数据库。
JDBC连接器hibernate.*透传属性
Kafka Connect支持透传配置使您能够通过从连接器配置直接传递某些属性来修改底层系统的行为。默认情况下一些Hibernate属性通过JDBC连接器的连接属性例如connection.url、connection.username和connection.pool.*_size以及连接器的运行时属性例如database.time_zone、quote.identifiers进行公开。
如果您想自定义其他Hibernate行为可以利用透传机制将使用hibernate.*命名空间的属性添加到连接器配置中。例如为了帮助Hibernate解析目标数据库的类型和版本可以添加hibernate.dialect属性并将其设置为数据库的完全限定类名例如org.hibernate.dialect.MariaDBDialect。
常见问题
是否需要使用ExtractNewRecordState单一消息转换
不需要这实际上是Debezium JDBC连接器与其他实现的区别之一。虽然连接器能够摄取类似竞争对手的扁平化事件但它还可以本地地摄取Debezium的复杂变更事件结构而无需任何特定类型的转换。
如果更改列的类型或者重命名或删除列是否由模式演变处理
不Debezium JDBC连接器不会对现有列进行任何更改。连接器支持的模式演变非常基础。它只是将事件结构中的字段与表的列列表进行比较然后添加尚未在表中定义为列的字段。如果列的类型或默认值发生更改连接器不会在目标数据库中进行调整。如果列被重命名旧列保持不变连接器将在表中追加一个具有新名称的列但是旧列中的现有数据行保持不变。此类模式更改应手动处理。
如果列的类型未解析为我想要的类型如何强制映射到不同的数据类型
Debezium JDBC连接器使用复杂的类型系统来解析列的数据类型。有关此类型系统如何将特定字段的模式定义解析为JDBC类型的详细信息请参阅数据和列类型映射部分。如果要应用不同的数据类型映射请手动定义表以明确获取首选的列类型。
如何在不更改Kafka主题名称的情况下指定表名的前缀或后缀
为了向目标表名添加前缀或后缀调整table.name.format连接器配置属性以应用所需的前缀或后缀。例如要为所有表名添加jdbc_前缀将table.name.format配置属性的值设置为jdbc_${topic}。如果连接器订阅了一个名为orders的主题那么生成的表将被创建为jdbc_orders。
为什么有些列会自动加引号即使没有启用标识符引用
在某些情况下即使没有启用quote.identifiers特定的列或表名可能仍会被显式引用。这通常是在列或表名以特定约定开头或使用特定约定时否则会被视为非法语法。例如当primary.key.mode设置为kafka时某些数据库只允许以下划线开头的列名在列名被引用时才有效。引用行为是特定于方言的并在不同类型的数据库之间有所不同。
更多Debezium技术内容
更多Debezium技术请参考
Debezium技术专栏
