山东网站制作,游戏推广平台有哪些,订单网站模板,dedecms 股票网站模板Debezium日常分享系列之#xff1a;Debezium2.5稳定版本之Mysql连接器的工作原理 一、Mysql连接器的工作原理1.支持的 MySQL 拓扑2.MariaDB 支持3.Schema history topic4.Schema change topic5.Snapshots1#xff09;使用全局读锁的初始快照2#xff09;Debezium MySQL 连接… Debezium日常分享系列之Debezium2.5稳定版本之Mysql连接器的工作原理 一、Mysql连接器的工作原理1.支持的 MySQL 拓扑2.MariaDB 支持3.Schema history topic4.Schema change topic5.Snapshots1使用全局读锁的初始快照2Debezium MySQL 连接器用于执行具有全局读锁的初始快照的默认工作流程3使用表级锁的初始快照4Debezium MySQL 连接器用于执行带有表级锁的初始快照的默认工作流程5了解为什么初始快照捕获所有表的架构历史记录6从初始快照未捕获的表中捕获数据无架构更改7从初始快照未捕获的表中捕获数据架构更改 6.临时快照1触发临时增量快照2触发临时阻塞快照 7.增量快照1增量快照流程2Debezium 如何解决具有相同主键的记录之间的冲突3快照窗口4触发增量快照5具有附加条件的临时增量快照6使用Kafka信令通道触发增量快照7具有附加条件的临时增量快照8停止增量快照9使用Kafka信令通道停止增量快照10只读增量快照11即席只读增量快照12快照事件操作类型 8.阻止快照1阻塞快照进程2配置快照 9.可能重复10.主题名称11.事务元数据12.变更数据事件丰富 二、总结 MySQL 有一个二进制日志binlog它按照提交到数据库的顺序记录所有操作。这包括对表架构的更改以及对表中数据的更改。 MySQL 使用 binlog 进行复制和恢复。
Debezium MySQL 连接器读取 binlog为行级 INSERT、UPDATE 和 DELETE 操作生成更改事件并将更改事件发送到 Kafka 主题。客户端应用程序读取这些 Kafka 主题。
由于 MySQL 通常设置为在指定时间段后清除二进制日志因此 MySQL 连接器会对每个数据库执行初始一致快照。 MySQL 连接器从创建快照的位置读取二进制日志。
一、Mysql连接器的工作原理
连接器支持的 MySQL 拓扑概述对于规划您的应用程序非常有用。为了以最佳方式配置和运行 Debezium MySQL 连接器了解连接器如何跟踪表结构、公开架构更改、执行快照以及确定 Kafka 主题名称会很有帮助。
注意Debezium MySQL 连接器已经过测试并正式支持 MariaDB 11.1.2。有关如何启用官方 MariaDB 支持模式的详细信息请参阅 MariaDB 支持部分。
1.支持的 MySQL 拓扑
Debezium MySQL 连接器支持以下 MySQL 拓扑
Standalone 当使用单个 MySQL 服务器时服务器必须启用 binlog并且可以选择启用 GTID以便 Debezium MySQL 连接器可以监控服务器。这通常是可以接受的因为二进制日志也可以用作增量备份。在这种情况下MySQL 连接器始终连接到并遵循此独立 MySQL 服务器实例。 Primary and replica Debezium MySQL 连接器可以跟踪主服务器之一或副本之一如果该副本启用了二进制日志但连接器只能看到该服务器可见的集群中的更改。一般来说除了多主拓扑之外这不是问题。连接器在服务器的 binlog 中记录其位置该位置在集群中的每台服务器上都不同。因此连接器必须仅跟随一个 MySQL 服务器实例。如果该服务器发生故障则必须重新启动或恢复该服务器然后连接器才能继续。 High available clusters MySQL 有多种高可用性解决方案它们使 MySQL 更容易容忍并且几乎可以立即从问题和故障中恢复。大多数 HA MySQL 集群都使用 GTID以便副本能够跟踪任何主服务器上的所有更改。 Multi-primary 网络数据库 (NDB) 集群复制使用一个或多个 MySQL 副本节点每个副本节点都从多个主服务器进行复制。这是聚合多个 MySQL 集群的复制的强大方法。此拓扑需要使用 GTID。Debezium MySQL 连接器可以使用这些多主 MySQL 副本作为源并且只要新副本追上旧副本就可以故障转移到不同的多主 MySQL 副本。也就是说新副本具有第一个副本上看到的所有事务。即使连接器仅使用数据库和/或表的子集此功能也有效因为连接器可以配置为在尝试重新连接到新的多主 MySQL 副本并在数据库中找到正确位置时包含或排除特定 GTID 源。二进制日志。 Hosted Debezium MySQL 连接器支持使用 Amazon RDS 和 Amazon Aurora 等托管选项。由于这些托管选项不允许全局读锁因此使用表级锁来创建一致快照。
2.MariaDB 支持
虽然可以使用 MySQL 驱动程序连接并传输来自 MariaDB 的更改但我们建议将 Debezium MySQL 连接器配置为使用 MariaDB 适配器模式从而允许您利用 MariaDB 驱动程序及其独特的功能堆栈。要切换 MariaDB 支持模式必须使用 mariadb 值指定connector.adapter 配置属性。此模式使用 MariaDB 驱动程序而不是 MySQL 驱动程序这意味着您还必须提供与 MariaDB 兼容的数据库协议和 JDBC 驱动程序字符串请参见下面的示例。
MariaDB 补充配置
{...connector.adapter: mariadb,database.protocol: jdbc:mariadb,database.jdbc.driver: org.mariadb.jdbc.Driver
}注意当新的 Debezium MariaDB 连接器在未来版本中可用时将不再需要这些补充配置。由于连接器当前包含 MySQL 和 MariaDB 驱动程序因此在 Debezium 引入新的 MariaDB 特定连接器之前MariaDB 模式是可选的。
通过此配置Debezium MySQL 连接器将使用新的 MariaDB 适配器连接器该连接器本机流式传输 MariaDB 二进制事务日志中的更改。所有其他 MySQL 连接器属性都可以与上述补充配置结合使用。
3.Schema history topic
当数据库客户端查询数据库时客户端使用数据库的当前架构。但是数据库架构可以随时更改这意味着连接器必须能够识别记录每个插入、更新或删除操作时的架构。此外连接器不一定将当前架构应用于每个事件。如果事件相对较旧则它可能是在应用当前模式之前记录的。为了确保正确处理架构更改后发生的事件MySQL 在事务日志中不仅包含影响数据的行级更改还包含应用于数据库的 DDL 语句。当连接器在 binlog 中遇到这些 DDL 语句时它会解析它们并更新每个表架构的内存中表示。连接器使用此架构表示来识别每次插入、更新或删除操作时的表结构并生成适当的更改事件。在单独的数据库架构历史 Kafka 主题中连接器记录所有 DDL 语句以及每个 DDL 语句在 binlog 中出现的位置。当连接器在崩溃或正常停止后重新启动时它会从特定位置即特定时间点开始读取二进制日志。连接器通过读取数据库模式历史 Kafka 主题并解析直到连接器启动的 binlog 中的所有 DDL 语句来重建此时存在的表结构。此数据库架构历史记录主题仅供内部连接器使用。 可选连接器还可以将架构更改事件发送到面向消费者应用程序的不同主题。当 MySQL 连接器捕获应用了架构更改工具例如 gh-ost 或 pt-online-schema-change的表中的更改时会在迁移过程中创建帮助程序表。您必须配置连接器以捕获这些帮助器表中发生的更改。如果使用者不需要连接器为帮助程序表生成的记录请配置单个消息转换 (SMT) 以从连接器发出的消息中删除这些记录。
4.Schema change topic
您可以配置 Debezium MySQL 连接器来生成架构更改事件这些事件描述应用于数据库中的表的架构更改。连接器将架构更改事件写入名为 的 Kafka 主题其中 topicPrefix 是 topic.prefix 连接器配置属性中指定的命名空间。连接器发送到架构更改主题的消息包含有效负载并且可选还包含更改事件消息的架构。
架构更改事件的架构具有以下元素
name架构更改事件消息的名称。type更改事件消息的类型。version架构的版本。版本是一个整数每次架构更改时都会递增。fields更改事件消息中包含的字段。
示例MySQL 连接器架构更改主题的架构 以下示例显示了 JSON 格式的典型架构。
{schema: {type: struct,fields: [{type: string,optional: false,field: databaseName}],optional: false,name: io.debezium.connector.mysql.SchemaChangeKey,version: 1},payload: {databaseName: inventory}
}架构更改事件消息的负载包括以下元素
ddl提供导致架构更改的 SQL CREATE、ALTER 或 DROP 语句。databaseName应用 DDL 语句的数据库的名称。 databaseName 的值用作消息键。pos语句在二进制日志中出现的位置。tableChanges架构更改后整个表架构的结构化表示。 tableChanges 字段包含一个数组其中包含表中每一列的条目。由于结构化表示以 JSON 或 Avro 格式呈现数据因此消费者可以轻松读取消息而无需先通过 DDL 解析器进行处理。
重要
对于处于捕获模式的表连接器不仅将架构更改历史记录存储在架构更改主题中而且还存储在内部数据库架构历史主题中。内部数据库架构历史记录主题仅供连接器使用不适合消费应用程序直接使用。确保需要有关架构更改的通知的应用程序仅使用来自架构更改主题的信息。切勿对数据库架构历史主题进行分区。为了使数据库架构历史主题正常运行它必须维护连接器向其发出的事件记录的一致的全局顺序。为了确保主题不会在分区之间拆分请使用以下方法之一设置主题的分区计数 如果您手动创建数据库架构历史主题请将分区计数指定为 1。如果您使用 Apache Kafka 代理自动创建数据库架构历史主题则创建该主题时请将 Kafka num.partitions 配置选项的值设置为 1。
注意
连接器向其架构更改主题发出的消息格式处于孵化状态如有更改恕不另行通知。
示例发送到 MySQL 连接器架构更改主题的消息
以下示例显示了 JSON 格式的典型架构更改消息。该消息包含表模式的逻辑表示。
{schema: { },payload: {source: { version: 2.5.2.Final,connector: mysql,name: mysql,ts_ms: 1651535750218, snapshot: false,db: inventory,sequence: null,table: customers,server_id: 223344,gtid: null,file: mysql-bin.000003,pos: 570,row: 0,thread: null,query: null},databaseName: inventory, schemaName: null,ddl: ALTER TABLE customers ADD middle_name varchar(255) AFTER first_name, tableChanges: [ {type: ALTER, id: \inventory\.\customers\, table: { defaultCharsetName: utf8mb4,primaryKeyColumnNames: [ id],columns: [ {name: id,jdbcType: 4,nativeType: null,typeName: INT,typeExpression: INT,charsetName: null,length: null,scale: null,position: 1,optional: false,autoIncremented: true,generated: true},{name: first_name,jdbcType: 12,nativeType: null,typeName: VARCHAR,typeExpression: VARCHAR,charsetName: utf8mb4,length: 255,scale: null,position: 2,optional: false,autoIncremented: false,generated: false},{name: middle_name,jdbcType: 12,nativeType: null,typeName: VARCHAR,typeExpression: VARCHAR,charsetName: utf8mb4,length: 255,scale: null,position: 3,optional: true,autoIncremented: false,generated: false},{name: last_name,jdbcType: 12,nativeType: null,typeName: VARCHAR,typeExpression: VARCHAR,charsetName: utf8mb4,length: 255,scale: null,position: 4,optional: false,autoIncremented: false,generated: false},{name: email,jdbcType: 12,nativeType: null,typeName: VARCHAR,typeExpression: VARCHAR,charsetName: utf8mb4,length: 255,scale: null,position: 5,optional: false,autoIncremented: false,generated: false}],attributes: [ {customAttribute: attributeValue}]}}]}
}表 1. 发送到架构更改主题的消息中的字段描述
ItemField nameDescription1source源字段的结构与连接器写入表特定主题的标准数据更改事件完全相同。该字段对于关联不同主题的事件很有用。2ts_ms显示连接器处理事件的时间的可选字段。该时间基于运行 Kafka Connect 任务的 JVM 中的系统时钟。在源对象中ts_ms 表示数据库中进行更改的时间。通过将payload.source.ts_ms的值与payload.ts_ms的值进行比较您可以确定源数据库更新与Debezium之间的滞后。3databaseName、schemaName标识包含更改的数据库和架构。 databaseName 字段的值用作记录的消息键。4ddl该字段包含负责架构更改的 DDL。 ddl字段可以包含多个DDL语句。每个语句都适用于databaseName 字段中的数据库。多个 DDL 语句按照它们应用于数据库的顺序出现。客户端可以提交适用于多个数据库的多个 DDL 语句。如果 MySQL 以原子方式应用它们连接器将按顺序获取 DDL 语句按数据库对它们进行分组并为每个组创建一个架构更改事件。如果 MySQL 单独应用它们连接器会为每个语句创建一个单独的架构更改事件。5tableChanges包含由 DDL 命令生成的架构更改的一项或多项的数组。6type描述了这种变化。该值为以下之一CREATE:表已创建。ALTER:表已修改。DROP:表已删除。7id创建、更改或删除的表的完整标识符。在表重命名的情况下该标识符是 、 表名称的串联。8table表示应用更改后的表元数据。9primaryKeyColumnNames组成表主键的列的列表。10columns已更改表中每列的元数据。11attributes每个表更改的自定义属性元数据。
5.Snapshots
当 Debezium MySQL 连接器首次启动时它会执行数据库的初始一致快照。此快照使连接器能够为数据库的当前状态建立基线。
Debezium 在运行快照时可以使用不同的模式。快照模式由 snapshot.mode 配置属性确定。该属性的默认值为初始值。您可以通过更改 snapshot.mode 属性的值来自定义连接器创建快照的方式。
连接器在执行快照时完成一系列任务。确切的步骤因快照模式和对数据库有效的表锁定策略而异。 Debezium MySQL 连接器在执行使用全局读锁或表级锁的初始快照时完成不同的步骤。
1使用全局读锁的初始快照
您可以通过更改 snapshot.mode 属性的值来自定义连接器创建快照的方式。如果您配置不同的快照模式连接器将使用此工作流的修改版本来完成快照。有关不允许全局读锁的环境中的快照过程的信息请参阅表级锁的快照工作流程。
2Debezium MySQL 连接器用于执行具有全局读锁的初始快照的默认工作流程
下表显示了 Debezium 创建具有全局读锁的快照所遵循的工作流程中的步骤。
步骤行动1建立与数据库的连接。2确定要捕获的表。默认情况下连接器捕获所有非系统表的数据。快照完成后连接器将继续传输指定表的数据。如果您希望连接器仅从特定表捕获数据则可以通过设置 table.include.list 或 table.exclude.list 等属性指示连接器仅捕获表或表元素子集的数据。3在要捕获的表上获取全局读锁以阻止其他数据库客户端的写入。快照本身不会阻止其他客户端应用可能干扰连接器尝试读取二进制日志位置和表架构的 DDL。连接器在读取二进制日志位置时保留全局读锁并按后续步骤中所述释放锁。4使用可重复读取语义启动事务以确保事务中的所有后续读取都是针对一致快照完成的。注意使用这些隔离语义可能会减慢快照的进度。如果快照需要很长时间才能完成请考虑使用不同的隔离配置或者跳过初始快照并运行增量快照。5读取当前的binlog位置。6捕获数据库中所有表的结构或指定捕获的所有表的结构。连接器将架构信息保留在其内部数据库架构历史主题中包括所有必要的 DROP… 和 CREATE… DDL 语句。架构历史记录提供有关发生更改事件时有效的结构的信息。注意默认情况下连接器捕获数据库中每个表的架构包括未配置为捕获的表。如果表未配置为捕获则初始快照仅捕获其结构它不捕获任何表数据。有关快照为何保留未包含在初始快照中的表的架构信息的更多信息请参阅了解为何初始快照捕获所有表的架构。7释放步骤3中获得的全局读锁。其他数据库客户端现在可以写入数据库。8在连接器在步骤 5 中读取的二进制日志位置连接器开始扫描指定用于捕获的表。在扫描期间连接器完成以下任务1.确认该表是在快照开始之前创建的。如果该表是在快照开始后创建的则连接器会跳过该表。快照完成并且连接器转换为流式传输后它会为快照开始后创建的任何表发出更改事件。2.为从表中捕获的每一行生成一个读取事件。所有读取事件都包含相同的二进制日志位置即在步骤 5 中获得的位置。3.将每个读取事件发送到源表的 Kafka 主题。4.释放数据表锁如果适用。9提交交易。10在连接器偏移量中记录快照的成功完成。
生成的初始快照捕获捕获表中每行的当前状态。从该基线状态开始连接器会捕获发生的后续更改。
快照流程开始后如果由于连接器故障、重新平衡或其他原因导致该流程中断则该流程将在连接器重新启动后重新启动。
连接器完成初始快照后它会继续从步骤 5 中读取的位置进行流式传输以便不会错过任何更新。
如果连接器因任何原因再次停止则在重新启动后它将从之前停止的位置恢复流式传输更改。
连接器重新启动后如果日志已被修剪则连接器在日志中的位置可能不再可用。然后连接器失败并返回一个错误指示需要新快照。要将连接器配置为在这种情况下自动启动快照请将 snapshot.mode 属性的值设置为when_needed。有关 Debezium MySQL 连接器故障排除的更多提示请参阅出现问题时的行为。
3使用表级锁的初始快照
在某些数据库环境中管理员不允许全局读锁。如果 Debezium MySQL 连接器检测到不允许全局读锁则连接器在执行快照时将使用表级锁。为了使连接器执行使用表级锁的快照Debezium 连接器用于连接 MySQL 的数据库帐户必须具有 LOCK TABLES 权限。
4Debezium MySQL 连接器用于执行带有表级锁的初始快照的默认工作流程
以下工作流程列出了 Debezium 创建具有表级读锁的快照所采取的步骤。有关不允许全局读锁的环境中的快照过程的信息请参阅全局读锁的快照工作流程。
步骤行动1建立与数据库的连接。2确定要捕获的表。默认情况下连接器捕获所有非系统表。要让连接器捕获表或表元素的子集您可以设置多个包含和排除属性来过滤数据例如 table.include.list 或 table.exclude.list。3获取表级锁。4使用可重复读取语义启动事务以确保事务中的所有后续读取都是针对一致快照完成的。5读取当前的binlog位置。6读取连接器配置为捕获更改的数据库和表的架构。连接器将架构信息保留在其内部数据库架构历史主题中包括所有必要的 DROP… 和 CREATE… DDL 语句。架构历史记录提供有关发生更改事件时有效的结构的信息。注意默认情况下连接器捕获数据库中每个表的架构包括未配置为捕获的表。如果表未配置为捕获则初始快照仅捕获其结构它不捕获任何表数据。有关快照为何保留未包含在初始快照中的表的架构信息的更多信息请参阅了解为何初始快照捕获所有表的架构。7在连接器在步骤 5 中读取的二进制日志位置连接器开始扫描指定用于捕获的表。在扫描期间连接器完成以下任务1.确认该表是在快照开始之前创建的。如果该表是在快照开始后创建的则连接器会跳过该表。快照完成并且连接器转换为流式传输后它会为快照开始后创建的任何表发出更改事件。2.为从表中捕获的每一行生成一个读取事件。所有读取事件都包含相同的二进制日志位置即在步骤 5 中获得的位置。3.将每个读取事件发送到源表的 Kafka 主题。4.释放数据表锁如果适用。8提交交易。9释放表级锁。其他数据库客户端现在可以写入任何先前锁定的表。10在连接器偏移量中记录快照的成功完成。
5了解为什么初始快照捕获所有表的架构历史记录
连接器运行的初始快照捕获两种类型的信息
表数据有关连接器的 table.include.list 属性中命名的表中的 INSERT、UPDATE 和 DELETE 操作的信息。
模式数据描述应用于表的结构更改的 DDL 语句。架构数据将持久保存到内部架构历史主题和连接器的架构更改主题如果已配置。
运行初始快照后您可能会注意到快照捕获未指定捕获的表的架构信息。默认情况下初始快照旨在捕获数据库中存在的每个表的架构信息而不仅仅是从指定捕获的表中捕获。连接器要求表的架构存在于架构历史记录主题中然后才能捕获表。通过启用初始快照来捕获不属于原始捕获集的表的架构数据Debezium 使连接器做好准备以便在以后有必要时可以轻松地从这些表中捕获事件数据。如果初始快照未捕获表的架构则必须先将该架构添加到历史主题然后连接器才能从表中捕获数据。
在某些情况下您可能希望限制初始快照中的架构捕获。当您想要减少完成快照所需的时间时这会很有用。或者当 Debezium 通过有权访问多个逻辑数据库的用户帐户连接到数据库实例但您希望连接器仅捕获特定逻辑数据库中的表中的更改时。
附加信息、
从初始快照未捕获的表中捕获数据无架构更改从初始快照未捕获的表中捕获数据架构更改设置 schema.history.internal.store.only.captured.tables.ddl 属性以指定要从中捕获架构信息的表。设置 schema.history.internal.store.only.captured.databases.ddl 属性以指定从中捕获架构更改的逻辑数据库。
6从初始快照未捕获的表中捕获数据无架构更改
在某些情况下您可能希望连接器从初始快照未捕获其架构的表中捕获数据。根据连接器配置初始快照可能仅捕获数据库中特定表的表架构。如果历史主题中不存在表架构连接器将无法捕获该表并报告缺少架构错误。
您可能仍然能够从表中捕获数据但必须执行其他步骤来添加表架构。
先决条件
您想要从具有连接器在初始快照期间未捕获的架构的表中捕获数据。在事务日志中表的所有条目都使用相同的架构。有关从发生结构更改的新表中捕获数据的信息请参阅从初始快照未捕获的表中捕获数据架构更改。
程序
1.停止连接器。2.删除 schema.history.internal.kafka.topic 属性指定的内部数据库架构历史记录主题。3.对连接器配置应用以下更改 a.将 snapshot.mode 设置为 schema_only_recovery。b.将 schema.history.internal.store.only.captured.tables.ddl 的值设置为 false。c.将您希望连接器捕获的表添加到 table.include.list。这保证了连接器将来可以重建所有表的架构历史记录。 4.重新启动连接器。快照恢复过程根据表的当前结构重建架构历史记录。5.可选快照完成后启动增量快照以捕获新添加的表的现有数据以及该连接器离线时发生的其他表的更改。6.可选将 snapshot.mode 重置回 schema_only以防止连接器在将来重新启动后启动恢复。
7从初始快照未捕获的表中捕获数据架构更改
如果将架构更改应用于表则架构更改之前提交的记录与更改之后提交的记录具有不同的结构。当 Debezium 从表中捕获数据时它会读取架构历史记录以确保将正确的架构应用于每个事件。如果该架构不存在于架构历史记录主题中则连接器无法捕获该表并会产生错误。
如果要从初始快照未捕获的表中捕获数据并且该表的架构已修改则必须将该架构添加到历史主题如果尚不可用。您可以通过运行新的架构快照或运行表的初始快照来添加架构。
先决条件
您想要从具有连接器在初始快照期间未捕获的架构的表中捕获数据。对表应用了架构更改以便要捕获的记录不具有统一的结构。
程序
初始快照捕获了所有表的架构store.only.captured.tables.ddl 设置为 false
1.编辑 table.include.list 属性以指定要捕获的表。2.重新启动连接器。3.如果您想从新添加的表中捕获现有数据请启动增量快照。
初始快照未捕获所有表的架构store.only.captured.tables.ddl 设置为 true 如果初始快照未保存要捕获的表的架构请完成以下过程之一
流程 1架构快照然后是增量快照
在此过程中连接器首先执行架构快照。然后您可以启动增量快照以使连接器能够同步数据。
1.停止连接器。2.删除 schema.history.internal.kafka.topic 属性指定的内部数据库架构历史记录主题。3.清除配置的Kafka Connect offset.storage.topic中的偏移量。注意删除偏移量只能由具有操作内部 Kafka Connect 数据经验的高级用户执行。此操作具有潜在的破坏性只能作为最后的手段执行。4.按照以下步骤所述设置连接器配置中的属性值 a.将 snapshot.mode 属性的值设置为 schema_only。b.编辑 table.include.list 以添加要捕获的表。 5.重新启动连接器。6.等待 Debezium 捕获新表和现有表的架构。连接器停止后任何表发生的数据更改都不会被捕获。7.为了保证数据不丢失需要启动增量快照。
程序 2初始快照然后是可选的增量快照
在此过程中连接器执行数据库的完整初始快照。与任何初始快照一样在具有许多大型表的数据库中运行初始快照可能是一项耗时的操作。快照完成后您可以选择触发增量快照以捕获连接器离线时发生的任何更改。
1.停止连接器。2.删除 schema.history.internal.kafka.topic 属性指定的内部数据库架构历史记录主题。3.清除配置的Kafka Connect offset.storage.topic中的偏移量。注意删除偏移量只能由具有操作内部 Kafka Connect 数据经验的高级用户执行。此操作具有潜在的破坏性只能作为最后的手段执行。4.编辑 table.include.list 以添加要捕获的表。5.按照以下步骤所述设置连接器配置中的属性值 将 snapshot.mode 属性的值设置为初始值。可选将 schema.history.internal.store.only.captured.tables.ddl 设置为 false。 6.重新启动连接器。连接器拍摄完整的数据库快照。快照完成后连接器将转换为流式传输。可选要捕获连接器离线时更改的任何数据请启动增量快照。
6.临时快照
默认情况下连接器仅在首次启动后运行初始快照操作。在此初始快照之后在正常情况下连接器不会重复快照过程。连接器捕获的任何未来更改事件数据仅通过流处理传入。
但是在某些情况下连接器在初始快照期间获取的数据可能会过时、丢失或不完整。为了提供重新捕获表数据的机制Debezium 包含一个执行临时快照的选项。在 Debezium 环境中发生以下任何更改后您可能需要执行临时快照
修改连接器配置以捕获一组不同的表。Kafka 主题已删除必须重建。由于配置错误或其他一些问题会发生数据损坏。
您可以通过启动所谓的临时快照为之前捕获快照的表重新运行快照。特别快照需要使用信令表。您可以通过向 Debezium 信令表发送信号请求来启动临时快照。
当您启动现有表的临时快照时连接器会将内容追加到该表已存在的主题中。如果删除了先前存在的主题并且启用了自动主题创建Debezium 可以自动创建主题。
即席快照信号指定要包含在快照中的表。快照可以捕获数据库的全部内容也可以仅捕获数据库中表的子集。此外快照可以捕获数据库中表的内容的子集。
您可以通过向信令表发送执行快照消息来指定要捕获的表。将执行快照信号的类型设置为增量或阻塞并提供要包含在快照中的表的名称如下表所述
表 2. 即席执行快照信号记录示例
字段默认值描述typeincremental指定要运行的快照的类型。目前您可以请求增量快照或阻塞快照。data-collectionsN/A包含与要快照的表的完全限定名称匹配的正则表达式的数组。名称的格式与 signal.data.collection 配置选项相同。additional-conditionN/A可选字符串指定基于表的列的条件以捕获表内容的子集。注意该属性已被弃用。要指定用于定义要快照捕获的数据子集的条件请使用additional-conditions 参数。additional-conditionsN/A一个可选数组指定连接器评估的一组附加条件以确定要包含在快照中的记录子集。每个附加条件都是一个对象指定过滤临时快照捕获的数据的条件。您可以为每个附加条件设置以下参数data-collection过滤器应用到的表的完全限定名称。您可以对每个表应用不同的过滤器。filter指定数据库记录中必须存在的列值快照才能包含该列值例如“color‘blue’”。您分配给过滤器参数的值与您在为阻塞快照设置 snapshot.select.statement.overrides 属性时可能在 SELECT 语句的 WHERE 子句中指定的值类型相同。在早期 Debezium 版本中没有为快照信号定义显式过滤器参数相反过滤条件是由为现已弃用的附加条件参数指定的值隐含的。surrogate-keyN/A一个可选字符串指定连接器在快照过程中用作表主键的列名称。
1触发临时增量快照
您可以通过向信令表添加具有执行快照信号类型的条目来启动临时增量快照。连接器处理消息后开始快照操作。快照进程读取第一个和最后一个主键值并将这些值用作每个表的起点和终点。根据表中的条目数和配置的块大小Debezium 将表划分为块并继续对每个块进行快照一次一个。
2触发临时阻塞快照
您可以通过向信令表添加具有执行快照信号类型的条目来启动临时阻塞快照。连接器处理消息后开始快照操作。连接器暂时停止流式传输然后启动指定表的快照遵循初始快照期间使用的相同过程。快照完成后连接器将恢复流式传输。
7.增量快照
为了提供管理快照的灵活性Debezium 包含一个补充快照机制称为增量快照。增量快照依赖 Debezium 机制向 Debezium 连接器发送信号。增量快照基于DDD-3设计文档。
在增量快照中Debezium 不像初始快照那样一次性捕获数据库的完整状态而是以一系列可配置块的形式分阶段捕获每个表。您可以指定希望快照捕获的表以及每个块的大小。块大小确定快照在数据库上的每个提取操作期间收集的行数。增量快照的默认块大小为 1024 行。
随着增量快照的进行Debezium 使用水印来跟踪其进度维护其捕获的每个表行的记录。与标准初始快照过程相比这种分阶段捕获数据的方法具有以下优点
您可以与流数据捕获并行运行增量快照而不是推迟流数据直到快照完成。连接器在整个快照过程中持续从更改日志中捕获近乎实时的事件并且两个操作都不会阻塞另一个操作。如果增量快照的进度中断您可以恢复增量快照而不会丢失任何数据。进程恢复后快照将从停止点开始而不是从头开始重新捕获表。您可以随时按需运行增量快照并根据需要重复该过程以适应数据库更新。例如您可以在修改连接器配置以将表添加到其 table.include.list 属性后重新运行快照。
1增量快照流程
当您运行增量快照时Debezium 按主键对每个表进行排序然后根据配置的块大小将表拆分为块。逐块工作然后捕获块中的每个表行。对于它捕获的每一行快照都会发出一个 READ 事件。该事件表示块快照开始时行的值。
随着快照的进行其他进程可能会继续访问数据库从而可能修改表记录。为了反映此类更改INSERT、UPDATE 或 DELETE 操作将照常提交到事务日志。同样正在进行的 Debezium 流处理继续检测这些更改事件并将相应的更改事件记录发送到 Kafka。
2Debezium 如何解决具有相同主键的记录之间的冲突
在某些情况下流处理发出的 UPDATE 或 DELETE 事件的接收顺序不正确。也就是说在快照捕获包含该行的 READ 事件的块之前流处理可能会发出一个修改表行的事件。当快照最终发出该行相应的 READ 事件时其值已被取代。为了确保按正确的逻辑顺序处理不按顺序到达的增量快照事件Debezium 采用缓冲方案来解决冲突。仅当快照事件和流式事件之间的冲突得到解决后Debezium 才会向 Kafka 发送事件记录。
3快照窗口
为了帮助解决迟到的 READ 事件和修改同一表行的流式事件之间的冲突Debezium 采用了所谓的快照窗口。快照窗口划定了增量快照捕获指定表块数据的时间间隔。在块的快照窗口打开之前Debezium 会遵循其通常的行为并将事件从事务日志直接向下游发送到目标 Kafka 主题。但从特定块的快照打开的那一刻起直到其关闭Debezium 都会执行重复数据删除步骤来解决具有相同主键的事件之间的冲突。
对于每个数据收集Debezium 会发出两种类型的事件并将它们的记录存储在单个目标 Kafka 主题中。它直接从表中捕获的快照记录作为 READ 操作发出。同时随着用户继续更新数据集合中的记录并且更新事务日志以反映每次提交Debezium 会针对每次更改发出 UPDATE 或 DELETE 操作。
当快照窗口打开时Debezium 开始处理快照块它将快照记录传送到内存缓冲区。在快照窗口期间缓冲区中 READ 事件的主键与传入流事件的主键进行比较。如果未找到匹配项则流式事件记录将直接发送到 Kafka。如果 Debezium 检测到匹配它会丢弃缓冲的 READ 事件并将流式记录写入目标主题因为流式事件在逻辑上取代静态快照事件。块的快照窗口关闭后缓冲区仅包含不存在相关事务日志事件的 READ 事件。 Debezium 将这些剩余的 READ 事件发送到表的 Kafka 主题。
连接器对每个快照块重复该过程。
4触发增量快照
目前启动增量快照的唯一方法是将临时快照信号发送到源数据库上的信令表。
您将信号作为 SQL INSERT 查询提交到信令表。
Debezium 检测到信令表中的更改后它会读取信号并运行请求的快照操作。
您提交的查询指定要包含在快照中的表并且可以选择指定快照操作的类型。目前快照操作的唯一有效选项是默认值增量。
要指定要包含在快照中的表请提供一个列出表的数据集合数组或用于匹配表的正则表达式数组例如
{data-collections: [public.MyFirstTable, public.MySecondTable]}增量快照信号的数据收集数组没有默认值。如果数据收集数组为空Debezium 会检测到不需要执行任何操作并且不会执行快照。
注意
如果要包含在快照中的表的名称在数据库、架构或表的名称中包含点 (.)则要将该表添加到数据集合数组中必须转义该表的每个部分名称用双引号引起来。例如要包含公共架构中存在且名称为 My.Table 的表请使用以下格式“public”.“My.Table”。
先决条件
信令已启用。 源数据库中存在信令数据集合。信令数据收集在 signal.data.collection 属性中指定。
使用源信令通道触发增量快照
1.发送 SQL 查询以将即席增量快照请求添加到信令表
INSERT INTO signalTable (id, type, data) VALUES (id, snapshotType, {data-collections: [tableName,tableName],type:snapshotType,additional-conditions:[{data-collection: tableName, filter: additional-condition}]});例如
INSERT INTO myschema.debezium_signal (id, type, data)
values (ad-hoc-1, execute-snapshot, {data-collections: [schema1.table1, schema2.table2], type:incremental, additional-conditions:[{data-collection: schema1.table1 ,filter:color\blue\}]}); 命令中的id、type、data参数的取值与信令表的字段相对应。
示例中的参数说明如下表
表 3. 用于向信令表发送增量快照信号的 SQL 命令中的字段说明
序列值描述1myschema.debezium_signal指定源数据库上信令表的完全限定名称。2ad-hoc-1id 参数指定指定为信号请求的 id 标识符的任意字符串。使用此字符串来标识信令表中条目的日志消息。 Debezium 不使用该字符串。相反在快照期间Debezium 会生成自己的 id 字符串作为水印信号。3execute-snapshot类型参数指定信号要触发的操作。4data-collections信号数据字段的必需组件指定表名称或正则表达式的数组以匹配要包含在快照中的表名称。该数组列出了按完全限定名称匹配表的正则表达式使用与在 signal.data.collection 配置属性中指定连接器信令表名称相同的格式。5incremental信号数据字段的可选类型组件指定要运行的快照操作的类型。目前唯一有效的选项是默认值增量。如果不指定值连接器将运行增量快照。6additional-conditions一个可选数组指定连接器评估的一组附加条件以确定要包含在快照中的记录子集。每个附加条件都是一个具有数据收集和过滤属性的对象。您可以为每个数据集合指定不同的过滤器。data-collection 属性是要应用过滤器的数据集合的完全限定名称。有关附加条件参数的更多信息请参阅具有附加条件的临时增量快照。
5具有附加条件的临时增量快照
如果您希望快照仅包含表中内容的子集您可以通过向快照信号附加附加条件参数来修改信号请求。
典型快照的 SQL 查询采用以下形式
SELECT * FROM tableName ....通过添加附加条件参数您可以将 WHERE 条件附加到 SQL 查询如下例所示
SELECT * FROM data-collection WHERE filter ....以下示例显示了一个 SQL 查询用于将带有附加条件的临时增量快照请求发送到信令表
INSERT INTO signalTable (id, type, data) VALUES (id, snapshotType, {data-collections: [tableName,tableName],type:snapshotType,additional-conditions:[{data-collection: tableName, filter: additional-condition}]});例如假设您有一个包含以下列的产品表
id (primary key)colorquantity
如果您希望products表的增量快照只包含colorblue的数据项可以使用以下SQL语句触发快照
INSERT INTO myschema.debezium_signal (id, type, data) VALUES(ad-hoc-1, execute-snapshot, {data-collections: [schema1.products],type:incremental, additional-conditions:[{data-collection: schema1.products, filter: colorblue}]});附加条件参数还允许您传递基于多个列的条件。例如使用上一示例中的产品表您可以提交一个查询来触发增量快照该快照仅包含那些 colorblue 且数量10 的商品的数据
INSERT INTO myschema.debezium_signal (id, type, data) VALUES(ad-hoc-1, execute-snapshot, {data-collections: [schema1.products],type:incremental, additional-conditions:[{data-collection: schema1.products, filter: colorblue AND quantity10}]});以下示例显示了连接器捕获的增量快照事件的 JSON。
示例增量快照事件消息
{before:null,after: {pk:1,value:New data},source: {...snapshot:incremental 1},op:r, 2ts_ms:1620393591654,transaction:null
}选项字段名描述1snapshot指定要运行的快照操作的类型。目前唯一有效的选项是默认值增量。在提交到信令表的 SQL 查询中指定类型值是可选的。如果不指定值连接器将运行增量快照。2op指定事件类型。快照事件的值为r表示READ操作。
6使用Kafka信令通道触发增量快照
您可以向配置的 Kafka 主题发送消息请求连接器运行临时增量快照。
Kafka 消息的键必须与 topic.prefix 连接器配置选项的值匹配。
消息的值是一个带有类型和数据字段的 JSON 对象。
信号类型为execute-snapshot数据字段必须有以下字段
表 4. 执行快照数据字段
字段默认值描述typeincremental要执行的快照的类型。目前 Debezium 仅支持增量类型。data-collectionsN/A一组以逗号分隔的正则表达式与要包含在快照中的表的完全限定名称相匹配。使用与 signal.data.collection 配置选项所需的格式相同的格式指定名称。additional-conditionN/A一个可选字符串指定连接器评估的条件以指定要包含在快照中的记录子集。注意此属性已弃用应由附加条件属性替换。additional-conditionsN/A附加条件的可选数组指定连接器评估的条件以指定要包含在快照中的记录子集。每个附加条件都是一个对象指定过滤临时快照捕获的数据的条件。您可以为每个附加条件设置以下参数 data-collection:: 过滤器应用到的表的完全限定名称。您可以对每个表应用不同的过滤器。 filter:: 指定数据库记录中必须存在的列值快照才能包含该列值例如“color‘blue’”。您分配给筛选器参数的值与您在为阻塞快照设置 snapshot.select.statement.overrides 属性时可能在 SELECT 语句的 WHERE 子句中指定的值类型相同。在早期 Debezium 版本中没有为快照信号定义显式过滤器参数相反过滤条件是由为现已弃用的附加条件参数指定的值隐含的。
执行快照 Kafka 消息的示例
Key test_connectorValue {type:execute-snapshot,data: {data-collections: [schema1.table1, schema1.table2], type: INCREMENTAL}}
7具有附加条件的临时增量快照
Debezium 使用附加条件字段来选择表内容的子集。
通常当 Debezium 运行快照时它会运行 SQL 查询例如
SELECT * FROM tableName ….当快照请求包含附加条件属性时该属性的数据收集和过滤参数将附加到 SQL 查询中例如
SELECT * FROM data-collection WHERE filter ….例如给定一个具有 id主键、颜色和品牌列的产品表如果您希望快照仅包含 color‘blue’ 的内容则当您请求快照时您可以添加附加 -用于过滤内容的条件属性
Key test_connectorValue {type:execute-snapshot,data: {data-collections: [schema1.products], type: INCREMENTAL, additional-conditions: [{data-collection: schema1.products ,filter:colorblue}]}}您可以使用additional-conditions 属性来传递基于多列的条件。例如使用与上例相同的产品表如果您希望快照仅包含产品表中 color‘blue’ 和 Brand‘MyBrand’ 的内容您可以发送以下请求
Key test_connectorValue {type:execute-snapshot,data: {data-collections: [schema1.products], type: INCREMENTAL, additional-conditions: [{data-collection: schema1.products ,filter:colorblue AND brandMyBrand}]}}8停止增量快照
您还可以通过向源数据库上的表发送信号来停止增量快照。您可以通过发送 SQL INSERT 查询向表提交停止快照信号。
Debezium 检测到信令表中的变化后会读取信号并停止正在进行的增量快照操作。
您提交的查询指定增量快照操作并且可以选择删除当前运行快照的表。
先决条件
信令已启用。 源数据库中存在信令数据集合。信令数据收集在 signal.data.collection 属性中指定。
使用源信令通道停止增量快照
1.发送 SQL 查询以停止到信令表的即席增量快照
INSERT INTO signalTable (id, type, data) values (id, stop-snapshot, {data-collections: [tableName,tableName],type:incremental});例如
INSERT INTO myschema.debezium_signal (id, type, data)
values (ad-hoc-1, stop-snapshot, {data-collections: [schema1.table1, schema2.table2], type:incremental}); signal命令中的id、type、data参数的取值与信令表的字段相对应。
示例中的参数说明如下表
表 5. 用于向信令表发送停止增量快照信号的 SQL 命令中的字段说明
序列值描述1myschema.debezium_signal指定源数据库上信令表的完全限定名称。2ad-hoc-1id 参数指定指定为信号请求的 id 标识符的任意字符串。使用此字符串来标识信令表中条目的日志消息。 Debezium 不使用该字符串。3stop-snapshot指定类型参数指定信号要触发的操作。4data-collections信号数据字段的可选组件指定表名称或正则表达式的数组以匹配要从快照中删除的表名称。该数组列出了按完全限定名称匹配表的正则表达式使用与在 signal.data.collection 配置属性中指定连接器信令表名称相同的格式。如果省略数据字段的该组成部分则该信号将停止正在进行的整个增量快照。5incremental信号数据字段的必需组成部分指定要停止的快照操作类型。目前唯一有效的选项是增量选项。如果不指定类型值则信号无法停止增量快照。
9使用Kafka信令通道停止增量快照
您可以向配置的 Kafka 信令主题发送信号消息以停止即席增量快照。
Kafka 消息的键必须与 topic.prefix 连接器配置选项的值匹配。
消息的值是一个带有类型和数据字段的 JSON 对象。
信号类型为stop-snapshot数据字段必须有以下字段
表 6. 执行快照数据字段
字段值描述typeincremental要执行的快照的类型。目前 Debezium 仅支持增量类型。data-collectionsN/A一个可选的逗号分隔正则表达式数组与要包含在快照中的表的完全限定名称相匹配。使用与 signal.data.collection 配置选项所需的格式相同的格式指定名称。
以下示例显示了典型的停止快照 Kafka 消息
Key test_connectorValue {type:stop-snapshot,data: {data-collections: [schema1.table1, schema1.table2], type: INCREMENTAL}}10只读增量快照
MySQL 连接器允许通过与数据库的只读连接运行增量快照。要运行具有只读访问权限的增量快照连接器使用设置为高水位线和低水位线的已执行全局事务 ID (GTID)。通过将二进制日志 (binlog) 事件的 GTID 或服务器的心跳与低水位线和高水位线进行比较来更新块窗口的状态。
要切换到只读实现请将 read.only 属性的值设置为 true。
先决条件
启用 MySQL GTID。如果连接器从多线程副本即replica_parallel_workers 值大于 0 的副本读取数据则必须设置以下选项之一 replica_preserve_commit_orderONSlave_preserve_commit_orderON
11即席只读增量快照
当 MySQL 连接为只读时您可以使用任何可用的信令通道而无需使用源通道。
12快照事件操作类型
MySQL 连接器将快照事件作为 READ 操作“op”“r”发出。如果您希望连接器将快照事件作为 CREATE © 事件发出请配置 Debezium ReadToInsertEvent 单消息转换 (SMT) 以修改事件类型。
以下示例显示如何配置 SMT
示例使用 ReadToInsertEvent SMT 更改快照事件的类型
transformssnapshotasinsert,...
transforms.snapshotasinsert.typeio.debezium.connector.mysql.transforms.ReadToInsertEvent8.阻止快照
为了在管理快照方面提供更大的灵活性Debezium 包含一个补充的临时快照机制称为阻塞快照。阻止快照依赖 Debezium 机制向 Debezium 连接器发送信号。
阻塞快照的行为就像初始快照一样只是您可以在运行时触发它。
在以下情况下您可能希望运行阻塞快照而不是使用标准初始快照进程
您添加一个新表并希望在连接器运行时完成快照。您添加了一个大型表并且希望快照在比增量快照更短的时间内完成。
1阻塞快照进程
当您运行阻塞快照时Debezium 会停止流式传输然后启动指定表的快照遵循与初始快照期间使用的相同流程。快照完成后流将恢复。
2配置快照
您可以在信号的数据组件中设置以下属性
data-collections指定哪些表必须是快照附加条件您可以为不同的表指定不同的过滤器。 data-collection 属性是要应用过滤器的表的完全限定名称。过滤器属性将具有与 snapshot.select.statement.overrides 中使用的相同值
例如
{type: blocking, data-collections: [schema1.table1, schema1.table2], additional-conditions: [{data-collection: schema1.table1, filter: SELECT * FROM [schema1].[table1] WHERE column1 0 ORDER BY column2 DESC}, {data-collection: schema1.table2, filter: SELECT * FROM [schema1].[table2] WHERE column2 0}]}9.可能重复
发送信号以触发快照的时间与流停止和快照开始的时间之间可能存在延迟。由于此延迟在快照完成后连接器可能会发出一些与快照捕获的记录重复的事件记录。
10.主题名称
默认情况下MySQL 连接器将表中发生的所有 INSERT、UPDATE 和 DELETE 操作的更改事件写入特定于该表的单个 Apache Kafka 主题。
连接器使用以下约定来命名更改事件主题
topicPrefix.databaseName.tableName
假设fulfillment 是主题前缀inventory 是数据库名称并且数据库包含名为orders、customers 和products 的表。 Debezium MySQL 连接器向三个 Kafka 主题发出事件每个主题对应数据库中的每个表
fulfillment.inventory.orders
fulfillment.inventory.customers
fulfillment.inventory.products以下列表提供了默认名称的组件的定义
主题前缀由 topic.prefix 连接器配置属性指定的主题前缀。架构名称发生操作的模式的名称。表名发生操作的表的名称。
连接器应用类似的命名约定来标记其内部数据库架构历史主题、架构更改主题和事务元数据主题。
如果默认的主题名称不能满足您的需求您可以配置自定义主题名称。要配置自定义主题名称您可以在逻辑主题路由 SMT 中指定正则表达式。
11.事务元数据
Debezium 可以生成代表事务边界并丰富数据更改事件消息的事件。
注意
Debezium 接收事务元数据的时间限制Debezium 仅注册和接收部署连接器后发生的事务的元数据。部署连接器之前发生的事务的元数据不可用。
Debezium 为每个事务中的 BEGIN 和 END 分隔符生成事务边界事件。事务边界事件包含以下字段
statusBEGIN or ENDid唯一交易标识符的字符串表示形式。ts_ms数据源处事务边界事件BEGIN 或 END 事件的时间。如果数据源未向 Debezium 提供事件时间则该字段表示 Debezium 处理事件的时间。event_count (for END events)事务发出的事件总数。data_collections (for END events)data_collection 和 event_count 元素对的数组指示连接器针对源自数据集合的更改发出的事件数。
例子
{status: BEGIN,id: 0e4d5dcd-a33b-11ea-80f1-02010a22a99e:10,ts_ms: 1486500577125,event_count: null,data_collections: null
}{status: END,id: 0e4d5dcd-a33b-11ea-80f1-02010a22a99e:10,ts_ms: 1486500577691,event_count: 2,data_collections: [{data_collection: s1.a,event_count: 1},{data_collection: s2.a,event_count: 1}]
}除非通过 topic.transaction 选项覆盖否则连接器会将事务事件发送到 topic.prefix.transaction 主题。
12.变更数据事件丰富
启用事务元数据后数据消息信封将通过新的事务字段进行丰富。该字段以字段组合的形式提供有关每个事件的信息
id唯一交易标识符的字符串表示形式。total_order该事件在事务生成的所有事件中的绝对位置。data_collection_order事件在事务发出的所有事件中的每个数据收集位置。
以下是消息示例
{before: null,after: {pk: 2,aa: 1},source: {
...},op: c,ts_ms: 1580390884335,transaction: {id: 0e4d5dcd-a33b-11ea-80f1-02010a22a99e:10,total_order: 1,data_collection_order: 1}
}对于未启用 GTID 的系统事务标识符是使用 binlog 文件名和 binlog 位置的组合构建的。例如如果事务 BEGIN 事件对应的 binlog 文件名和位置分别为 mysql-bin.000002 和 1913则 Debezium 构造的事务标识符将为 filemysql-bin.000002,pos1913。
二、总结
更多Debezium技术请参考
Debezium专栏
