定制的卡夫卡自动主题创建连接

卡夫卡为自动创建主题提供了两种机制。您可以启用自动主题创建卡夫卡的经纪公司,从卡夫卡2.6.0开始,你也可以使卡夫卡连接创建主题。卡夫卡代理使用auto.create.topics.enable属性来控制自动主题创建。在卡夫卡连接topic.creation.enable属性指定是否允许卡夫卡连接来创建主题。在这两种情况下,默认设置为允许自动主题创建属性。

启用自动主题创建时,如果Debezium源连接器发出更改事件纪录表没有目标主题已经存在,在运行开云体育官方注册网址时创建的主题是摄取到卡夫卡作为事件记录。

自动主题创建之间的差异在卡夫卡的代理和连接

主题代理创建共享一个默认的配置是有限的。代理不能应用配置不同的主题或集独特的主题。相比之下,卡夫卡连接可以申请任何的几个配置在创建主题,设置复制因子,分区数量,以及其他特定主题设置中指定Debezium连接器配置。开云体育官方注册网址连接器配置定义了一组主题创建组,和同事一组主题与每组配置属性。

代理配置和卡夫卡连接配置是相互独立的。卡夫卡连接可以创建主题无论你禁用主题创建代理。如果你启用自动主题创建代理和卡夫卡连接,连接配置优先,和代理创建主题只有在没有设置在卡夫卡连接配置应用。

卡夫卡的禁用自动主题创建代理

默认情况下,卡夫卡代理配置使代理能够在运行时创建主题如果主题不存在。主题由代理不能配置自定义属性。如果你使用比2.6.0卡夫卡版本,和你想创建主题与特定的配置,您必须禁用自动主题创建代理,然后明确创建主题,手动,或通过一个自定义部署过程。

过程

  • 在代理配置,设置的值auto.create.topics.enable来假。

卡夫卡连接设置

在卡夫卡自动主题创建连接的控制topic.creation.enable财产。属性的默认值真正的,使自动创建主题,如以下示例所示:

topic.creation.enable= true

的设置topic.creation.enable连接属性适用于所有工人的集群。

卡夫卡连接自动主题创建需要定义卡夫卡的配置属性连接适用于创建主题。您指定Debezium连接器配置主题配置属性通过定义主题组,然后指定属性适用于每组。开云体育官方注册网址连接器配置定义了一个默认主题创建组,可以选择一个或多个自定义主题创建组。创建自定义主题团体使用主题名称列表模式指定的主题组织的设置申请。

详情关于卡夫卡连接匹配主题主题创建组,明白了主题创建组。有关如何配置属性的更多信息被分配到组,明白了主题创建组的配置属性。

默认情况下,主题,卡夫卡连接创建命名基于模式server.schema.table例如,dbserver.myschema.inventory。

如果你不想让卡夫卡自动主题创建连接,设置的值topic.creation.enable来假在卡夫卡连接配置(connect-distributed.properties文件或通过环境变量CONNECT_TOPIC_CREATION_ENABLE当使用开云体育官方注册网址Debezium卡夫卡连接的容器的形象)。

卡夫卡连接自动主题创建要求replication.factor和分区至少在设置属性默认的主题创建组。是有效的团体来获得所需的属性的值从默认值卡夫卡代理。

配置

对卡夫卡连接自动创建主题,它需要的信息从源连接器的配置属性应用在创建主题。您定义的属性控制主题为每个Debezium连接器创建的配置。开云体育官方注册网址作为卡夫卡连接创建主题对于连接器发出的事件记录,生成的主题从适用的组获得它们的配置。配置适用于连接器只发出的事件记录。

主题创建组

一组与一个主题相关的主题属性创建组。最低限度,你必须定义一个默认的主题创建组并指定它的配置属性。除此之外,您可以定义一个或多个自定义主题创建组和指定为每个独特的属性。

当你创建自定义主题创建组,定义每个小组的成员主题基于主题名称模式。您可以指定命名模式描述的主题从每组包括或排除。的包括和排除属性包含逗号分隔的列表定义主题名称的正则表达式模式。例如,如果你想让一群包括从字符串开始的所有主题dbserver1.inventory,设置它的值topic.creation.inventory.include财产dbserver1 \ \ .inventory \ \ *。

如果你指定包括和排除属性自定义主题组,排除规则优先考虑,并覆盖包含规则。

主题创建组的配置属性

的默认的相关主题创建组和每个自定义组是一组独特的配置属性。您可以配置一组包括的任何卡夫卡主题级配置属性。例如,您可以指定清理旧政策主题段,保留时间,或者是主题压缩类型对于一个话题组。您必须定义至少一个最小属性集来描述的配置要创建主题。

如果没有注册自定义组,或者包括模式对于任何注册组不匹配任何主题的名称创建,然后卡夫卡连接使用的配置默认的集团创建主题。

看到配置Debezium主题开云体育官方注册网址Debeziu开云体育官方注册网址m安装指南中一般的主题配置注意事项。

默认组配置

可以使用卡夫卡连接自动主题创建之前,您必须创建一个默认主题创建组和定义一个配置。默认主题创建组的配置应用于任何主题名称不匹配包括自定义主题的列表模式创建组。

过程

  • 定义属性topic.creation.default组,将它们添加到连接器配置JSON,如以下示例所示:

    {…“topic.creation.default.replication.factor”: 3,(1)“topic.creation.default.partitions”: 10(2)“topic.creation.default.cleanup.policy”:“紧凑”,(3):“topic.creation.default.compression.type lz4”(4)…}

    你可以包括任何卡夫卡主题级配置属性的配置默认的组。

表1。连接器配置默认的主题创建组
项 描述

1

topic.creation.default.replication.factor定义了主题由默认的复制因子组。
replication.factor是强制性的默认的但可选自定义组。自定义组将回落默认的如果不设置。使用集团的价值1使用代理卡夫卡的默认值。

2

topic.creation.default.partitions定义了主题分区的数量由默认组。
分区是强制性的默认的但可选自定义组。自定义组将回落默认的如果不设置。使用集团的价值1使用代理卡夫卡的默认值。

3

topic.creation.default.cleanup.policy映射到cleanup.policy财产的主题层次配置参数并定义了日志保留政策。

4

topic.creation.default.compression.type映射到compression.type财产的主题层次配置参数并定义消息如何被压缩在硬盘上。

自定义组跌回默认的组只有所需的设置replication.factor和分区属性。如果配置为一个自定义主题组离开其他属性定义,中指定的值默认的集团并不适用。

自定义组配置

您可以定义多个自定义主题组,每个国家都有自己的配置。

过程

  • 定义一个自定义主题组,添加一个topic.creation。< group_name >其中包括财产连接器JSON,列表的属性自定义组后的组名。

    下面的例子展示了示例配置库存和applicationlogs创建自定义主题组:

    {…(1)“topic.creation.inventory。包括": "dbserver1\\.inventory\\.*",(2)“topic.creation.inventory。分区": 20, "topic.creation.inventory.cleanup.policy": "compact", "topic.creation.inventory.delete.retention.ms": 7776000000,(3)“topic.creation.applicationlogs。包括": "dbserver1\\.logs\\.applog-.*",(4)“topic.creation.applicationlogs。排除": "dbserver1\\.logs\\.applog-old-.*",(5)“topic.creation.applicationlogs.replication。topic.creation.applicationlogs因素”:1。”分区": 20, "topic.creation.applicationlogs.cleanup.policy": "delete", "topic.creation.applicationlogs.retention.ms": 7776000000, "topic.creation.applicationlogs.compression.type": "lz4", ... }
表2。连接器配置自定义库存和applicationlogs主题创建组
项 描述

1

定义的配置库存组。
的replication.factor和分区属性是可选的自定义组。如果没有值设置,自定义团体回落的值集默认的组。将值设置为1使用价值是卡夫卡代理设置。

2

topic.creation.inventory.include定义了一个正则表达式来匹配所有话题开始dbserver1.inventory。。的配置定义的库存集团是仅适用于主题名称匹配指定的正则表达式。

3

定义的配置applicationlogs组。
的replication.factor和分区属性是可选的自定义组。如果没有值设置,自定义团体回落的值集默认的组。将值设置为1使用价值是卡夫卡代理设置。

4

topic.creation.applicationlogs.include定义了一个正则表达式来匹配所有话题开始dbserver1.logs.applog -。的配置定义的applicationlogs集团是仅适用于主题名称匹配指定的正则表达式。因为一个排除财产也为这群定义,匹配的主题包括正则表达式可能进一步限制排除财产。

5

topic.creation.applicationlogs.exclude定义了一个正则表达式来匹配所有话题开始dbserver1.logs.applog-old -。的配置定义的applicationlogs集团是仅适用于主题名称不匹配给定的正则表达式。因为一个包括财产也为这群定义,配置的applicationlogs集团是仅适用于指定主题名称相匹配包括正则表达式和做不匹配指定的排除正则表达式。

注册自定义组

在您指定的配置任何自定义主题创建组,注册组。

过程

  • 通过添加注册自定义组topic.creation.groups财产连接器JSON,并指定一个以逗号分隔的组。

    下面的例子注册自定义主题创建组库存和applicationlogs:

    {…“topic.creation。组”:“库存,applicationlogs”,…}
完成配置

下面的例子展示了一个配置,包括配置完成默认的对于一个话题,以及配置库存和一个applicationlogs创建自定义主题组:

例子:配置一个默认主题创建组和两个自定义组
{…“topic.creation.default.replication.factor”: 3,“topic.creation.default.partitions”: 10“topic.creation.default.cleanup.policy”:“紧凑”、“topic.creation.default.compression.type”:“lz4”、“topic.creation。组”:“库存,applicationlogs”、“topic.creation.inventory。包括”:“dbserver1 \ \ .inventory \ \ *”,“topic.creation.inventory。分区": 20, "topic.creation.inventory.cleanup.policy": "compact", "topic.creation.inventory.delete.retention.ms": 7776000000, "topic.creation.applicationlogs.include": "dbserver1\\.logs\\.applog-.*", "topic.creation.applicationlogs.exclude": "dbserver1\\.logs\\.applog-old-.*", "topic.creation.applicationlogs.replication.factor": 1, "topic.creation.applicationlogs.partitions": 20, "topic.creation.applicationlogs.cleanup.policy": "delete", "topic.creation.applicationlogs.retention.ms": 7776000000, "topic.creation.applicationlogs.compression.type": "lz4" }

额外的资源

有关主题的更多信息自动创建您可以看一看这些资源: