服装行业网站开发,硬件开发学什么专业,石家庄网站外包公司,网站建设服务费记入什么科目Debezium日常分享系列之#xff1a;Debezium2.5稳定版本之MySQL连接器配置示例和Connector参数详解 一、MySQL 连接器配置示例二、添加连接器配置三、连接器属性四、必须的连接器配置属性五、高级 MySQL 连接器配置属性六、Debezium 连接器数据库架构历史配置属性七、用于配置… Debezium日常分享系列之Debezium2.5稳定版本之MySQL连接器配置示例和Connector参数详解 一、MySQL 连接器配置示例二、添加连接器配置三、连接器属性四、必须的连接器配置属性五、高级 MySQL 连接器配置属性六、Debezium 连接器数据库架构历史配置属性七、用于配置生产者和消费者客户端的传递数据库架构历史属性八、Debezium 连接器 Kafka 信号配置属性九、Debezium 连接器传递信号 Kafka 消费者客户端配置属性十、Debezium 连接器接收器通知配置属性十一、Debezium 连接器直通数据库驱动程序配置属性十二、Debezium技术总结 一、MySQL 连接器配置示例
以下是连接器实例的配置示例该实例从 192.168.99.100 端口 3306 上的 MySQL 服务器捕获数据我们在逻辑上将其命名为 fullfillment。通常您可以通过设置连接器可用的配置属性在 JSON 文件中配置 Debezium MySQL 连接器。
您可以选择为数据库中的架构和表的子集生成事件。或者您可以忽略、屏蔽或截断包含敏感数据、大于指定大小或不需要的列。
{name: inventory-connector, 1config: {connector.class: io.debezium.connector.mysql.MySqlConnector, 2database.hostname: 192.168.99.100, 3database.port: 3306, 4database.user: debezium-user, 5database.password: debezium-user-pw, 6database.server.id: 184054, 7topic.prefix: fullfillment, 8database.include.list: inventory, 9schema.history.internal.kafka.bootstrap.servers: kafka:9092, 10schema.history.internal.kafka.topic: schemahistory.fullfillment, 11include.schema.changes: true 12}
}1.注册到 Kafka Connect 服务时的连接器名称。2.连接器的类名。3.MySQL 服务器地址。4.MySQL 服务器端口号。5.具有适当权限的 MySQL 用户。6.MySQL 用户的密码。7.连接器的唯一 ID。8.MySQL 服务器或集群的主题前缀。9.由指定服务器托管的数据库列表。10.连接器用于向数据库架构历史记录主题写入和恢复 DDL 语句的 Kafka 代理列表。11.数据库架构历史主题的名称。本主题仅供内部使用消费者不应使用。12.指定连接器是否应生成 DDL 更改事件并将其发送到履行架构更改主题以供使用者使用的标志。
您可以使用 POST 命令将此配置发送到正在运行的 Kafka Connect 服务。该服务记录配置并启动一个执行以下操作的连接器任务
连接到 MySQL 数据库。在捕获模式下读取表的更改数据表。流将事件记录更改为 Kafka 主题。
二、添加连接器配置
要开始运行 MySQL 连接器请配置连接器配置并将该配置添加到 Kafka Connect 集群。
先决条件
MySQL 设置为与 Debezium 连接器一起使用。Debezium MySQL 连接器已安装。
程序
为 MySQL 连接器创建配置。使用 Kafka Connect REST API 将该连接器配置添加到您的 Kafka Connect 集群。
结果
连接器启动后它会对连接器配置的 MySQL 数据库执行一致的快照。然后连接器开始生成行级操作的数据更改事件并将更改事件记录流式传输到 Kafka 主题。
三、连接器属性
Debezium MySQL 连接器具有许多配置属性您可以使用它们来为您的应用程序实现正确的连接器行为。许多属性都有默认值。有关属性的信息组织如下 所需的连接器配置属性 高级连接器配置属性 数据库架构历史连接器配置属性用于控制 Debezium 如何处理从数据库架构历史主题读取的事件。 传递数据库架构历史属性 控制数据库驱动程序行为的传递数据库驱动程序属性。
四、必须的连接器配置属性
除非有默认值否则需要以下配置属性。
属性默认值描述name无默认值连接器的唯一名称。尝试使用相同名称再次注册失败。所有 Kafka Connect 连接器都需要此属性。connector.class无默认值连接器的 Java 类的名称。始终为 MySQL 连接器指定 io.debezium.connector.mysql.MySqlConnector。tasks.max1应为此连接器创建的最大任务数。 MySQL 连接器始终使用单个任务因此不使用此值因此默认值始终可以接受。connector.adaptermysql要使用的连接器适配器模式。database.protocoljdbc:mysql驱动程序连接字符串使用 JDBC 协议来连接数据库。database.jdbc.drivercom.mysql.cj.jdbc.Driver要使用的驱动程序类名称。当使用连接器附带的替代驱动程序时这会很有用。database.hostname无默认值MySQL数据库服务器的IP地址或主机名。database.port3306MySQL 数据库服务器的整数端口号。database.user无默认值连接到 MySQL 数据库服务器时使用的 MySQL 用户的名称。database.password无默认值连接到 MySQL 数据库服务器时使用的密码。topic.prefix无默认值主题前缀为 Debezium 捕获更改的特定 MySQL 数据库服务器/集群提供命名空间。主题前缀在所有其他连接器中应该是唯一的因为它用作接收此连接器发出的事件的所有 Kafka 主题名称的前缀。数据库服务器逻辑名称中只能使用字母数字字符、连字符、点和下划线。注意不要更改此属性的值。如果您更改名称值则在重新启动后连接器不会继续向原始主题发出事件而是向名称基于新值的主题发出后续事件。连接器也无法恢复其数据库架构历史主题。database.server.id无默认值该数据库客户端的数字 ID在 MySQL 集群中当前运行的所有数据库进程中必须是唯一的。该连接器作为另一台服务器具有唯一的 ID加入 MySQL 数据库集群以便它可以读取 binlog。database.include.list空字符串可选的、以逗号分隔的正则表达式列表与要捕获更改的数据库的名称相匹配。连接器不会捕获名称不在database.include.list 中的任何数据库中的更改。默认情况下连接器捕获所有数据库中的更改。为了匹配数据库的名称Debezium 应用您指定为锚定正则表达式的正则表达式。也就是说指定的表达式与数据库的整个名称字符串进行匹配它与数据库名称中可能存在的子字符串不匹配。如果您在配置中包含此属性则不要同时设置database.exclude.list 属性。database.exclude.list空字符串可选的、以逗号分隔的正则表达式列表与您不想捕获其更改的数据库名称相匹配。连接器捕获名称不在database.exclude.list 中的任何数据库中的更改。为了匹配数据库的名称Debezium 应用您指定为锚定正则表达式的正则表达式。也就是说指定的表达式与数据库的整个名称字符串进行匹配它与数据库名称中可能存在的子字符串不匹配。如果您在配置中包含此属性则不要同时设置database.include.list 属性。table.include.list空字符串可选的、以逗号分隔的正则表达式列表与要捕获其更改的表的完全限定表标识符相匹配。连接器不会捕获未包含在 table.include.list 中的任何表中的更改。每个标识符的格式为databaseName.tableName。默认情况下连接器会捕获正在捕获更改的每个数据库中每个非系统表中的更改。为了匹配表的名称Debezium 应用您指定为锚定正则表达式的正则表达式。也就是说指定的表达式与表的整个名称字符串进行匹配它与表名中可能存在的子字符串不匹配。如果您在配置中包含此属性则不要同时设置 table.exclude.list 属性。column.exclude.list空字符串可选的、以逗号分隔的正则表达式列表与要从更改事件记录值中排除的列的完全限定名称相匹配。列的完全限定名称的格式为databaseName.tableName.columnName。为了匹配列的名称Debezium 应用您指定为锚定正则表达式的正则表达式。也就是说指定的表达式与列的整个名称字符串相匹配它与列名中可能存在的子字符串不匹配。如果您在配置中包含此属性则不要同时设置 column.include.list 属性。column.include.list空字符串可选的、以逗号分隔的正则表达式列表与要包含在更改事件记录值中的列的完全限定名称相匹配。列的完全限定名称的格式为databaseName.tableName.columnName。为了匹配列的名称Debezium 应用您指定为锚定正则表达式的正则表达式。也就是说指定的表达式与列的整个名称字符串相匹配它与列名中可能存在的子字符串不匹配。如果您在配置中包含此属性请不要设置column.exclude.list 属性。skip.messages.without.changefalse指定当包含的列没有变化时是否跳过发布消息。如果按照column.include.list 或column.exclude.list 属性包含的列没有变化这实际上会过滤消息。column.truncate.to.length.charsn/a可选的、以逗号分隔的正则表达式列表与基于字符的列的完全限定名称相匹配。如果要在一组列中的数据超过属性名称中长度指定的字符数时截断数据请设置此属性。将长度设置为正整数值例如column.truncate.to.20.chars。列的完全限定名称遵循以下格式databaseName.tableName.columnName。为了匹配列的名称Debezium 应用您指定为锚定正则表达式的正则表达式。也就是说指定的表达式与列的整个名称字符串相匹配该表达式与列名中可能存在的子字符串不匹配。您可以在单个配置中指定具有不同长度的多个属性。column.mask.with.length.charsn/a可选的、以逗号分隔的正则表达式列表与基于字符的列的完全限定名称相匹配。如果您希望连接器屏蔽一组列的值例如如果它们包含敏感数据请设置此属性。将 length 设置为正整数以将指定列中的数据替换为属性名称中长度指定的星号 (*) 字符数。将长度设置为 0零以将指定列中的数据替换为空字符串。列的完全限定名称遵循以下格式databaseName.tableName.columnName。为了匹配列的名称Debezium 应用您指定为锚定正则表达式的正则表达式。也就是说指定的表达式与列的整个名称字符串相匹配该表达式与列名中可能存在的子字符串不匹配。您可以在单个配置中指定具有不同长度的多个属性。column.mask.hash.hashAlgorithm.with.salt.salt; column.mask.hash.v2.hashAlgorithm.with.salt.saltn/a可选的、以逗号分隔的正则表达式列表与基于字符的列的完全限定名称相匹配。列的完全限定名称的格式为 databaseName.tableName.columnName。为了匹配列的名称Debezium 应用您指定为锚定正则表达式的正则表达式。也就是说指定的表达式与列的整个名称字符串相匹配该表达式与列名中可能存在的子字符串不匹配。在生成的更改事件记录中指定列的值将替换为假名。假名由应用指定的哈希算法和盐所产生的哈希值组成。基于所使用的散列函数可以维护引用完整性同时用假名替换列值。 在以下示例中CzQMA0cB5K 是随机选择的盐。column.mask.hash.SHA-256.with.salt.CzQMA0cB5K inventory.orders.customerName, inventory.shipment.customerName如有必要笔名会自动缩短至列的长度。连接器配置可以包含指定不同哈希算法和盐的多个属性。根据使用的哈希算法、选择的盐和实际数据集生成的数据集可能不会被完全屏蔽。如果值在不同的地方或系统中进行散列则应使用散列策略版本 2 以确保保真度。column.propagate.source.typen/a可选的、以逗号分隔的正则表达式列表与您希望连接器发出表示列元数据的额外参数的列的完全限定名称相匹配。设置此属性后连接器会将以下字段添加到事件记录的架构中__debezium.source.column.type__debezium.source.column.length__debezium.source.column.scale 这些参数分别传播列的原始类型名称和长度对于可变宽度类型。使连接器能够发出这些额外的数据可以帮助正确调整接收器数据库中特定数字或基于字符的列的大小。列的完全限定名称遵循以下格式之一databaseName.tableName.columnName 或databaseName.schemaName.tableName.columnName。为了匹配列的名称Debezium 应用您指定为锚定正则表达式的正则表达式。也就是说指定的表达式与列的整个名称字符串相匹配该表达式与列名中可能存在的子字符串不匹配。datatype.propagate.source.typen/a可选的、以逗号分隔的正则表达式列表指定为数据库中的列定义的数据类型的完全限定名称。设置此属性后对于具有匹配数据类型的列连接器会发出在其架构中包含以下额外字段的事件记录__debezium.source.column.type__debezium.source.column.length__debezium.source.column.scale 这些参数分别传播列的原始类型名称和长度对于可变宽度类型。使连接器能够发出这些额外的数据可以帮助正确调整接收器数据库中特定数字或基于字符的列的大小。列的完全限定名称遵循以下格式之一databaseName.tableName.typeName 或databaseName.schemaName.tableName.typeName。为了匹配数据类型的名称Debezium 应用您指定为锚定正则表达式的正则表达式。也就是说指定的表达式与数据类型的整个名称字符串进行匹配该表达式与类型名称中可能存在的子字符串不匹配。time.precision.modeadaptive_time_microseconds时间、日期和时间戳可以用不同类型的精度表示包括Adaptive_time_microseconds默认值完全按照数据库中的方式捕获日期、日期时间和时间戳值根据数据库列的类型使用毫秒、微秒或纳秒精度值但 TIME 类型字段除外这些字段始终捕获为微秒。adaptive已弃用根据数据库列的类型使用毫秒、微秒或纳秒精度值捕获与数据库中完全相同的时间和时间戳值。connect始终使用 Kafka Connect 的内置时间、日期和时间戳表示来表示时间和时间戳值无论数据库列的精度如何它们都使用毫秒精度。decimal.handling.modeprecise指定连接器应如何处理 DECIMAL 和 NUMERIC 列的值precise默认使用以二进制形式在更改事件中表示的 java.math.BigDecimal 值来精确表示它们。double 使用 double 值表示它们这可能会导致精度损失但更易于使用。string 将值编码为格式化字符串这很容易使用但有关真实类型的语义信息会丢失。bigint.unsigned.handling.modelong指定 BIGINT UNSIGNED 列应如何在更改事件中表示。可能的设置有long 使用 Java 的 long 来表示值这可能无法提供精度但对于消费者来说很容易使用。 long 通常是首选设置。Precision 使用 java.math.BigDecimal 来表示值这些值通过使用二进制表示形式和 Kafka Connect 的 org.apache.kafka.connect.data.Decimal 类型在更改事件中进行编码。当使用大于 2^63 的值时请使用此设置因为这些值无法使用 long 来传送。include.schema.changestrue布尔值指定连接器是否应将数据库架构中的更改发布到与数据库服务器 ID 名称相同的 Kafka 主题。每个架构更改都使用包含数据库名称且其值包含 DDL 语句的键来记录。这与连接器内部记录数据库模式历史记录的方式无关。include.schema.commentsfalse布尔值指定连接器是否应解析和发布元数据对象上的表和列注释。启用此选项将对内存使用产生影响。逻辑模式对象的数量和大小在很大程度上影响 Debezium 连接器消耗的内存量并且向每个连接器添加潜在的大型字符串数据可能会非常昂贵。include.queryfalse布尔值指定连接器是否应包含生成更改事件的原始 SQL 查询。如果将此选项设置为 true则还必须使用 binlog_rows_query_log_events 选项配置 MySQL或将 binlog_annotate_row_events 选项设置为 ON 的 MariaDB。当 include.query 为 true 时对于快照进程生成的事件不存在查询。将 include.query 设置为 true 可能会公开通过在更改事件中包含原始 SQL 语句而显式排除或屏蔽的表或字段。因此默认设置为 false。event.deserialization.failure.handling.modefail指定连接器在二进制日志事件反序列化期间应如何对异常做出反应。此选项已弃用请改用 event.processing.failure.handling.mode 选项。inconsistent.schema.handling.modefail指定连接器应如何对与内部架构表示中不存在的表相关的二进制日志事件做出反应。即内部表示与数据库不一致。fail会引发异常指示有问题的事件及其二进制日志偏移量并导致连接器停止。warn 记录有问题的事件及其二进制日志偏移量并跳过该事件。skip有问题的事件并且不记录任何内容。connect.timeout.ms30000一个正整数值指定此连接器在尝试连接到 MySQL 数据库服务器后超时之前应等待的最长时间以毫秒为单位。默认为 30 秒。gtid.source.includes无默认值与 GTID 集中的源 UUID 匹配的正则表达式的逗号分隔列表连接器使用它来查找 MySQL 服务器上的 binlog 位置。设置此属性后连接器仅使用其源 UUID 与指定包含模式之一匹配的 GTID 范围。为了匹配 GTID 的值Debezium 应用您指定为锚定正则表达式的正则表达式。也就是说指定的表达式与整个 UUID 字符串进行匹配它与 UUID 中可能存在的子字符串不匹配。如果您在配置中包含此属性则不要同时设置 gtid.source.excludes 属性。gtid.source.excludes无默认值以逗号分隔的正则表达式列表与连接器用来查找 MySQL 服务器上的 binlog 位置的 GTID 集中的源 UUID 相匹配。设置此属性后连接器仅使用源 UUID 与任何指定排除模式都不匹配的 GTID 范围。为了匹配 GTID 的值Debezium 应用您指定为锚定正则表达式的正则表达式。也就是说指定的表达式与整个 UUID 字符串进行匹配它与 UUID 中可能存在的子字符串不匹配。如果您在配置中包含此属性则不要同时设置 gtid.source.includes 属性。tombstones.on.deletetrue控制删除事件后是否发生逻辑删除事件。true - 删除操作由删除事件和后续逻辑删除事件表示。false - 仅发出删除事件。删除源记录后如果为主题启用了日志压缩则发出逻辑删除事件默认行为允许 Kafka 完全删除与已删除行的键相关的所有事件。message.key.columnsn/a表达式列表指定连接器用于形成发布到指定表的 Kafka 主题的更改事件记录的自定义消息键的列。默认情况下Debezium 使用表的主键列作为其发出的记录的消息键。您可以根据一列或多列配置自定义消息键以代替默认值或者为缺少主键的表指定键。要为表建立自定义消息键请列出该表然后列出要用作消息键的列。每个列表条目均采用以下格式fully-qualified_tableName:,keyColumn 要将表键基于多个列名请在列名之间插入逗号。每个完全限定的表名都是以下格式的正则表达式databaseName.tableName该属性可以包含多个表的条目。使用分号分隔列表中的表条目。以下示例设置表 inventory.customers 和purchase.orders 的消息键inventory.customers:pk1,pk2;(.*).purchaseorders:pk3,pk4对于表 inventory.customer列 pk1 和 pk2 被指定为消息键。对于任何数据库中的purchaseorders 表列pk3 和pk4 作为消息键。用于创建自定义消息键的列数没有限制。但是最好使用指定唯一键所需的最小数量。binary.handling.modebytes指定应如何在更改事件中表示二进制列例如 blob、binary、varbinary。可能的设置bytes 将二进制数据表示为字节数组。Base64 将二进制数据表示为 Base64 编码的字符串。base64-url-safe 将二进制数据表示为 base64-url-safe 编码的字符串。hex 将二进制数据表示为十六进制编码 (base16) 字符串。schema.name.adjustment.modenone指定应如何调整架构名称以与连接器使用的消息转换器兼容。可能的设置none 不应用任何调整。avro 将 Avro 类型名称中不能使用的字符替换为下划线。avro_unicode 将 Avro 类型名称中不能使用的下划线或字符替换为相应的 unicode如 _uxxxx。注意_是转义序列类似于Java中的反斜杠field.name.adjustment.modenone指定应如何调整字段名称以与连接器使用的消息转换器兼容。可能的设置none 不应用任何调整。avro 将 Avro 类型名称中不能使用的字符替换为下划线。avro_unicode 将 Avro 类型名称中不能使用的下划线或字符替换为相应的 unicode如 _uxxxx。注意_是转义序列类似于Java中的反斜杠
五、高级 MySQL 连接器配置属性
下表描述了高级 MySQL 连接器属性。这些属性的默认值很少需要更改。因此您不需要在连接器配置中指定它们。
表 24. MySQL 连接器高级配置属性的描述
属性默认值描述connect.keep.alivetrue一个布尔值指定是否应使用单独的线程来确保与 MySQL 服务器/集群的连接保持活动状态。converters无默认值枚举连接器可以使用的自定义转换器实例的符号名称的逗号分隔列表。例如布尔值。需要此属性才能使连接器能够使用自定义转换器。对于为连接器配置的每个转换器还必须添加 .type 属性该属性指定实现转换器接口的类的完全限定名称。 .type 属性使用以下格式converterSymbolicName.type例如布尔类型io.debezium.connector.mysql.converters.TinyIntOneToBooleanConverter如果您想进一步控制已配置转换器的行为您可以添加一个或多个配置参数以将值传递给转换器。要将这些附加配置参数与转换器关联请在参数名称前加上转换器的符号名称。例如要定义指定布尔转换器处理的列子集的选择器参数请添加以下属性boolean.selectordb1.table1.*, db1.table2.column1table.ignore.builtintrue一个布尔值指定是否应忽略内置系统表。无论表包含和排除列表如何这都适用。默认情况下系统表不会捕获其更改并且对任何系统表进行更改时不会生成任何事件。database.ssl.modepreferred指定是否使用加密连接。可能的设置有disabled指定使用未加密的连接。preferred如果服务器支持安全连接则建立加密连接。如果服务器不支持安全连接则回退到未加密的连接。required建立加密连接如果由于任何原因无法建立则失败。verify_ca行为与必需的类似但它还会根据配置的证书颁发机构 (CA) 证书验证服务器 TLS 证书如果服务器 TLS 证书与任何有效的 CA 证书都不匹配则会失败verify_identity行为类似于 verify_ca但还验证服务器证书是否与远程连接的主机匹配。database.ssl.keystore无默认值密钥存储文件的位置。这是可选的可用于客户端和 MySQL 服务器之间的双向身份验证。binlog.buffer.size0binlog 读取器使用的前瞻缓冲区的大小。默认设置 0 禁用缓冲。在特定情况下MySQL binlog 中可能包含由 ROLLBACK 语句完成的未提交数据。典型的示例是使用保存点或在单个事务中混合临时表和常规表更改。当检测到事务开始时Debezium 会尝试前滚 binlog 位置并查找 COMMIT 或 ROLLBACK以便确定是否从事务中流式传输更改。 binlog 缓冲区的大小定义了 Debezium 在搜索事务边界时可以缓冲的事务中最大更改数量。如果事务的大小大于缓冲区则 Debezium 必须倒回并重新读取流式传输时未放入缓冲区的事件。max.batch.size2048正整数值指定在此连接器的每次迭代期间应处理的每批事件的最大大小。默认为 2048。max.queue.size8192正整数值指定阻塞队列可以保存的最大记录数。当 Debezium 读取从数据库流式传输的事件时它会将事件放入阻塞队列中然后再将其写入 Kafka。如果连接器摄取消息的速度快于将消息写入 Kafka 的速度或者当 Kafka 不可用时阻塞队列可以为从数据库读取更改事件提供反压。当连接器定期记录偏移量时队列中保存的事件将被忽略。始终将 max.queue.size 的值设置为大于 max.batch.size 的值。max.queue.size.in.bytes0一个长整数值指定阻塞队列的最大容量以字节为单位。默认情况下没有为阻塞队列指定卷限制。要指定队列可以使用的字节数请将此属性设置为正 long 值。如果还设置了 max.queue.size则当队列大小达到任一属性指定的限制时写入队列将被阻止。例如如果设置 max.queue.size1000并且max.queue.size.in.bytes5000则在队列包含 1000 条记录之后或在队列中的记录量之后写入队列将被阻止达到5000字节。poll.interval.ms500正整数值指定连接器在开始处理一批事件之前应等待新更改事件出现的毫秒数。默认为 500 毫秒或 0.5 秒。snapshot.modeinitial指定连接器启动时运行快照的条件。可能的设置有initial仅当未记录逻辑服务器名称的偏移量时连接器才会运行快照。initial_only仅当逻辑服务器名称没有记录偏移量时连接器才运行快照然后停止即它不会从 binlog 中读取更改事件。when_needed只要连接器认为有必要它就会在启动时运行快照。也就是说当没有可用的偏移量时或者当先前记录的偏移量指定服务器中不可用的二进制日志位置或 GTID 时。never连接器从不使用快照。首次使用逻辑服务器名称启动时连接器从 binlog 的开头读取。请谨慎配置此行为。仅当保证 binlog 包含数据库的完整历史记录时它才有效。schema_onl连接器运行架构的快照而不是数据。当您不需要主题包含一致的数据快照但需要它们仅包含自连接器启动以来的更改时此设置非常有用。schema_only_recovery这是已捕获更改的连接器的恢复设置。当您重新启动连接器时此设置可以恢复损坏或丢失的数据库架构历史记录主题。您可以定期设置它来“清理”意外增长的数据库模式历史记录主题。数据库架构历史主题需要无限保留。snapshot.locking.modeminimal控制连接器是否持有全局 MySQL 读锁以及持有全局 MySQL 读锁的时间以防止在连接器执行快照时对数据库进行任何更新。可能的设置有minimal连接器仅对快照的初始部分持有全局读锁在此期间连接器读取数据库架构和其他元数据。快照中的剩余工作包括从每个表中选择所有行。连接器可以通过使用可重复读取事务以一致的方式执行此操作。即使不再持有全局读锁并且其他 MySQL 客户端正在更新数据库情况也是如此。minimal_percona连接器仅对快照的初始部分持有全局备份锁在此期间连接器读取数据库架构和其他元数据。快照中的剩余工作包括从每个表中选择所有行。连接器可以通过使用可重复读取事务以一致的方式执行此操作。即使不再持有全局备份锁并且其他 MySQL 客户端正在更新数据库情况也是如此。此模式不会将表刷新到磁盘不会被长时间运行的读取阻塞并且仅在 Percona Server 中可用。extended阻止快照期间的所有写入。如果存在正在提交 MySQL 从 REPEATABLE READ 语义中排除的操作的客户端请使用此设置。none防止连接器在快照期间获取任何表锁。虽然所有快照模式都允许使用此设置但当且仅当快照运行时没有发生架构更改时才可以安全使用。对于使用MyISAM引擎定义的表尽管设置了该属性但当MyISAM获取表锁时表仍然会被锁定。此行为与 InnoDB 引擎不同InnoDB 引擎获取行级锁。heartbeat.interval.ms0控制连接器向 Kafka 主题发送心跳消息的频率。默认行为是连接器不发送心跳消息。检测信号消息对于监视连接器是否正在从数据库接收更改事件非常有用。检测信号消息可能有助于减少连接器重新启动时需要重新发送的更改事件的数量。要发送心跳消息请将此属性设置为正整数该整数表示心跳消息之间的毫秒数。heartbeat.action.query无默认值指定当连接器发送检测信号消息时连接器在源数据库上执行的查询。例如这可以用于定期捕获源数据库中执行的 GTID 集的状态。 INSERT INTO gtid_history_table (select * from mysql.gtid_executed)database.initial.statements无默认值建立与数据库的 JDBC 连接而不是读取事务日志的连接时要执行的 SQL 语句的分号分隔列表。要将分号指定为 SQL 语句中的字符而不是分隔符请使用两个分号 (;。连接器可能自行决定建立 JDBC 连接因此该属性仅用于配置会话参数。它不用于执行 DML 语句。enable.time.adjustertrue布尔值指示连接器是否将 2 位数年份规范转换为 4 位数。当转换完全委托给数据库时设置为 false。MySQL 允许用户插入 2 位或 4 位数字的年份值。对于 2 位数的值该值将映射到 1970 - 2069 范围内的年份。默认行为是连接器执行转换。source.struct.versionv2Debezium 事件中源块的架构版本。 Debezium 0.10 对源块的结构进行了一些重大更改以便统一所有连接器的暴露结构。通过将此选项设置为 v1可以生成早期版本中使用的结构。但是不建议使用此设置并计划在未来的 Debezium 版本中删除此设置。skipped.operationst在流式传输过程中将跳过的以逗号分隔的操作类型列表。这些操作包括c 表示插入/创建、u 表示更新、d 表示删除、t 表示截断none 表示不跳过任何操作。默认情况下会跳过截断操作。signal.data.collection无默认值用于向连接器发送信号的数据集合的完全限定名称。使用以下格式指定集合名称数据库名称.表名称signal.enabled.channelssource为连接器启用的信令通道名称列表。默认情况下以下通道可用source、kafka、file、jmxnotification.enabled.channels无默认值为连接器启用的通知通道名称列表。默认情况下以下通道可用sink、log、jmxincremental.snapshot.allow.schema.changesfalse允许在增量快照期间更改架构。启用后连接器将在增量快照期间检测架构更改并重新选择当前块以避免锁定 DDL。请注意不支持对主键进行更改如果在增量快照期间执行更改可能会导致不正确的结果。另一个限制是如果架构更改仅影响列的默认值则在从 binlog 流处理 DDL 之前不会检测到更改。这不会影响快照事件的值但快照事件的架构可能具有过时的默认值。incremental.snapshot.chunk.size1024连接器在增量快照块期间获取并读入内存的最大行数。增加块大小可以提高效率因为快照运行的快照查询数量越大查询的数量就越少。然而较大的块大小也需要更多的内存来缓冲快照数据。将块大小调整为可在您的环境中提供最佳性能的值。incremental.snapshot.watermarking.strategyinsert_insert指定连接器在增量快照期间使用的水印机制以对增量快照可能捕获并在流恢复后重新捕获的事件进行重复数据删除。您可以指定以下选项之一insert_insert当您发送信号来启动增量快照时对于 Debezium 在快照期间读取的每个块它都会向信令数据集合写入一个条目以记录打开快照窗口的信号。快照完成后Debezium 会插入第二个条目记录关闭窗口的信号。insert_delete当您发送信号来启动增量快照时对于 Debezium 读取的每个块它都会向信令数据集合写入一个条目以记录打开快照窗口的信号。快照完成后该条目将被删除。不会为关闭快照窗口的信号创建条目。设置此选项可防止信令数据收集的快速增长。read.onlyfalse切换到替代增量快照水印实现以避免写入信号数据收集provide.transaction.metadatafalse确定连接器是否生成具有事务边界的事件并使用事务元数据丰富变更事件信封。如果您希望连接器执行此操作请指定 true。event.processing.failure.handling.modefail指定应如何处理事件处理期间的故障即遇到损坏的事件时。默认情况下fail模式会引发异常指示有问题的事件及其位置从而导致连接器停止。warn模式不会引发异常而是会记录有问题的事件及其位置并跳过该事件。ignore模式完全忽略有问题的事件不记录任何日志。topic.naming.strategyio.debezium.schema.DefaultTopicNamingStrategyTopicNamingStrategy 类的名称应用于确定数据更改、架构更改、事务、心跳事件等的主题名称默认为 DefaultTopicNamingStrategy。topic.delimiter.指定主题名称的分隔符默认为…topic.cache.size10000用于在有界并发哈希图中保存主题名称的大小。该缓存将有助于确定与给定数据集合对应的主题名称。topic.heartbeat.prefix__debezium-heartbeat控制连接器向其发送心跳消息的主题的名称。主题名称具有以下模式topic.heartbeat.prefix.topic.prefix例如如果主题前缀为fulfillment则默认主题名称为__debezium-heartbeat.fulfillment。topic.transactiontransaction控制连接器向其发送事务元数据消息的主题的名称。主题名称具有以下模式topic.prefix.topic.transaction例如如果主题前缀是fulfillment则默认主题名称是fulfillment.transaction。snapshot.max.threads1指定连接器在执行初始快照时使用的线程数。要启用并行初始快照请将该属性设置为大于 1 的值。在并行初始快照中连接器会同时处理多个表。该功能正在孵化中。snapshot.tables.order.by.row.countdisabled控制连接器在执行初始快照时处理表的顺序。指定以下选项之一descending连接器根据行数从最高到最低按顺序快照表。ascending连接器根据行数从最低到最高顺序对表进行快照。disabled连接器在执行初始快照时忽略行计数。custom.metric.tags无默认值自定义指标标签将接受键值对来自定义 MBean 对象名称该名称应附加在常规名称的末尾每个键代表 MBean 对象名称的一个标签相应的值将是该标签的值关键群岛。例如k1v1k2v2。errors.max.retries-1失败前可重试错误例如连接错误的最大重试次数-1 无限制0 禁用 0 重试次数。
六、Debezium 连接器数据库架构历史配置属性
Debezium 提供了一组 schema.history.internal.* 属性用于控制连接器如何与架构历史主题交互。
下表描述了用于配置 Debezium 连接器的schema.history.internal 属性。
表 25. 连接器数据库模式历史配置属性
属性默认值描述schema.history.internal.kafka.topic无默认值连接器存储数据库架构历史记录的 Kafka 主题的全名。schema.history.internal.kafka.bootstrap.servers无默认值连接器用于建立与 Kafka 集群的初始连接的主机/端口对列表。此连接用于检索连接器先前存储的数据库模式历史记录并用于写入从源数据库读取的每个 DDL 语句。每对都应指向 Kafka Connect 进程使用的同一个 Kafka 集群。schema.history.internal.kafka.recovery.poll.interval.ms100一个整数值指定连接器在启动/恢复期间轮询持久数据时应等待的最大毫秒数。默认值为 100 毫秒。schema.history.internal.kafka.query.timeout.ms3000一个整数值指定连接器在使用 Kafka 管理客户端获取集群信息时应等待的最大毫秒数。schema.history.internal.kafka.create.timeout.ms30000一个整数值指定连接器在使用 Kafka 管理客户端创建 kafka 历史主题时应等待的最大毫秒数。schema.history.internal.kafka.recovery.attempts100在连接器恢复因错误而失败之前连接器应尝试读取持久历史数据的最大次数。没有收到数据后等待的最长时间为recovery.attempts × recovery.poll.interval.ms。schema.history.internal.skip.unparseable.ddlfalse一个布尔值指定连接器是否应忽略格式错误或未知的数据库语句或停止处理以便人工修复问题。安全默认值是 false。应谨慎使用跳过因为在处理二进制日志时它可能会导致数据丢失或损坏。schema.history.internal.store.only.captured.tables.ddlfalse一个布尔值指定连接器是记录架构或数据库中所有表的架构结构还是仅记录指定捕获的表的架构结构。指定以下值之一false (default) 在数据库快照期间连接器记录数据库中所有非系统表的架构数据包括未指定捕获的表。最好保留默认设置。如果您稍后决定从最初未指定捕获的表中捕获更改则连接器可以轻松地开始从这些表中捕获数据因为它们的架构结构已存储在架构历史记录主题中。 Debezium 需要表的模式历史记录以便它可以识别更改事件发生时存在的结构。true在数据库快照期间连接器仅记录 Debezium 从中捕获更改事件的表的表模式。如果更改默认值并且稍后将连接器配置为从数据库中的其他表捕获数据则连接器将缺少从表捕获更改事件所需的架构信息。schema.history.internal.store.only.captured.databases.ddlfalse一个布尔值指定连接器是否记录数据库实例中所有逻辑数据库的架构结构。指定以下值之一true连接器仅记录逻辑数据库中的表的模式结构以及 Debezium 从中捕获更改事件的模式。false连接器记录所有逻辑数据库的模式结构。注意MySQL 连接器的默认值为 true
七、用于配置生产者和消费者客户端的传递数据库架构历史属性
Debezium 依赖 Kafka 生产者将模式更改写入数据库模式历史主题。同样当连接器启动时它依赖 Kafka 使用者从数据库架构历史主题中读取。您可以通过将值分配给一组以 schema.history.internal. Producer.* 和 schema.history.internal.consumer.* 前缀开头的传递配置属性来定义 Kafka 生产者和消费者客户端的配置。传递生产者和消费者数据库架构历史属性控制一系列行为例如这些客户端如何保护与 Kafka 代理的连接如以下示例所示
schema.history.internal.producer.security.protocolSSL
schema.history.internal.producer.ssl.keystore.location/var/private/ssl/kafka.server.keystore.jks
schema.history.internal.producer.ssl.keystore.passwordtest1234
schema.history.internal.producer.ssl.truststore.location/var/private/ssl/kafka.server.truststore.jks
schema.history.internal.producer.ssl.truststore.passwordtest1234
schema.history.internal.producer.ssl.key.passwordtest1234schema.history.internal.consumer.security.protocolSSL
schema.history.internal.consumer.ssl.keystore.location/var/private/ssl/kafka.server.keystore.jks
schema.history.internal.consumer.ssl.keystore.passwordtest1234
schema.history.internal.consumer.ssl.truststore.location/var/private/ssl/kafka.server.truststore.jks
schema.history.internal.consumer.ssl.truststore.passwordtest1234
schema.history.internal.consumer.ssl.key.passwordtest1234八、Debezium 连接器 Kafka 信号配置属性
Debezium 提供了一组 signal.* 属性用于控制连接器如何与 Kafka 信号主题交互。
下表描述了 Kafka 信号属性。
表 26. Kafka 信号配置属性
属性默认值描述signal.kafka.topictopic.prefix-signal连接器监视临时信号的 Kafka 主题的名称。注意如果禁用自动创建主题则必须手动创建所需的信令主题。需要一个信令主题来保留信号顺序。信令主题必须具有单个分区。signal.kafka.groupIdkafka-signalKafka 消费者使用的组 ID 的名称。signal.kafka.bootstrap.servers无默认值连接器用于建立与 Kafka 集群的初始连接的主机/端口对列表。每对都引用 Debezium Kafka Connect 进程使用的 Kafka 集群。signal.kafka.poll.timeout.ms100一个整数值指定连接器在轮询信号时等待的最大毫秒数。kafka.consumer.offset.commit.enabledfalse为信号主题启用偏移量提交以保证至少一次传递。如果禁用则仅处理消费者启动和运行时收到的信号。当消费者宕机时收到的任何信号都会丢失。
九、Debezium 连接器传递信号 Kafka 消费者客户端配置属性
Debezium 连接器提供信号 Kafka 消费者的直通配置。直通信号属性以前缀 Signals.consumer.* 开头。例如连接器将 signal.consumer.security.protocolSSL 等属性传递给 Kafka 使用者。
Debezium 在将属性传递给 Kafka 信号使用者之前会去除属性中的前缀。
十、Debezium 连接器接收器通知配置属性
下表描述了通知属性。
表 27. 接收器通知配置属性
属性默认值描述notification.sink.topic.name无默认值从 Debezium 接收通知的主题的名称。当您配置 notification.enabled.channels 属性以将接收器包含为启用的通知通道之一时需要此属性。
十一、Debezium 连接器直通数据库驱动程序配置属性
Debezium 连接器提供数据库驱动程序的直通配置。传递数据库属性以前缀 driver.* 开头。例如连接器将 driver.foobarfalse 等属性传递给 JDBC URL。
与数据库模式历史客户端的传递属性的情况一样Debezium 在将属性传递到数据库驱动程序之前会从属性中去除前缀。
十二、Debezium技术总结
更多Debezium技术请参考
Debezium技术专栏
