应该在同一个Kafka主题中放入几种事件类型吗？

Posted 2021-05-02 存储与大数据每周谈

采用Apache Kafka等流平台，有个很重要的问题是：将使用哪些主题？特别是，如果要将一堆不同的事件作为消息发布到Kafka，是将它们放在同一主题中，还是将它们拆分为不同的主题？

主题最重要的功能是允许使用者指定它想要使用的消息子集。在一个极端情况下，所有数据都放在一个主题中可不是一个好主意，因为这意味着消费者无法选择感兴趣的事件 - 给他们的只会是所有的内容。在另一个极端，拥有数百万个不同的主题也不是一个好事，因为Kafka中的每个主题都会消耗资源消耗，因此拥有大量的主题就会对性能不利。

实际上，从性能的角度来看，重要的是分区数量。但由于Kafka中的每个主题至少有一个分区，如果你有n个主题，那么就不可避免地至少有n个分区。不久之前，Jun Rao撰写了一篇博文，解释了拥有多个分区的成本（端到端延迟，文件描述符，内存开销，故障后的恢复时间）。根据经验，如果您关注延迟问题，您应该关注每个代理节点上的（数量级）数百个主题分区。如果每个节点有数万个甚至数千个分区，则延迟会受到影响。

该性能参数给设计主题的结构提供了一些指导：如果您发现自己有数千个主题，那么将一些细粒度，低吞吐量的主题合并到粗粒度主题中是明智之举，从而减少分区的扩散。

然而，性能问题并不是结束。在我看来，更重要的是您的主题结构的数据完整性和数据建模方面。我们将在本文的其余部分讨论这些内容。

主题=相同类型的事件的集合？

常见的想法（根据我所拥有的几个对话，并根据邮件列表）似乎是：将同类型的所有事件放在同一主题中，并针对不同的事件类型使用不同的主题。这种思路让人联想到关系数据库，其中表是具有相同类型（即同一组列）的记录的集合，因此我们在关系表和Kafka主题之间进行类比。

该融合模式的注册表本质上强化了这种模式，因为它鼓励你在主题中的所有消息使用相同的Avro模式。该模式可以在保持兼容性的同时进化（例如，通过添加可选字段），但最终所有消息都必须符合某种记录类型。在我们介绍了更多背景之后，我们将在后面的帖子中再次讨论这个问题。

对于某些类型的流数据（例如记录的活动事件），要求同一主题中的所有消息都符合相同的模式是有意义的。但是，有些人正在使用Kafka来实现更多类似数据库的目的，例如事件溯源，或者在微服务之间交换数据。在这种情况下，我相信，它定义一个主题为一组具有相同模式的消息并不重要。更重要的是Kafka维护主题分区内的消息排序。

这些事件的顺序很重要。例如，我们期望在客户做任何动作之前创建客户，并且我们也期望在客户关闭其帐户之后不再发生任何其他事情。使用Kafka时，您可以通过将它们全部放在同一个分区中来保留这些事件的顺序。在此示例中，您将使用客户ID作为分区键，然后将所有这些不同的事件放在同一主题中。它们必须位于同一主题中，因为不同的主题意味着不同的分区，并且不会跨分区保留排序。

排序问题

如果消费者暂停一段时间（可能是维护或部署新版本），则重新排序的风险尤其高。当消费者停止时，事件将继续发布，并且这些事件将存储在Kafka代理的选定主题分区中。当消费者再次启动时，它会消耗来自其所有输入分区的积压事件。如果消费者只有一个输入，那就没问题了：挂起的事件只是按照它们存储的顺序依次处理。但是，如果消费者有几个输入主题，它将选择输入主题以按任意顺序读取。它可以在读取另一个输入主题上的积压之前从一个输入主题读取所有挂起事件，或者它可以以某种方式交错输入。

因此，如果你把customerCreated，customerAddressChanged以及customerInvoicePaid事件在三个独立的主题，消费者可能会看到所有的customerAddressChanged事件，它看到任何之前customerCreated的事件。因此，消费者可能会看到一个customerAddressChanged客户的事件，根据其对世界的看法，尚未创建。

您可能想要为每条消息附加时间戳，并将其用于事件排序。如果要将事件导入数据仓库，您可以在事后对事件进行排序，这可能就可以了。但是在流进程中，时间戳是不够的：如果你得到一个具有特定时间戳的事件，你不知道你是否仍然需要等待一个时间戳较低的先前事件，或者所有之前的事件是否已到达而你是’准备好处理这个事件。依靠时钟同步通常会导致噩梦;

何时分割主题，何时结合？

鉴于这种背景，我将提出一些经验法则来帮助您确定在同一主题中放入哪些内容，以及将哪些内容拆分为单独的主题：

1. 最重要的规则是，  任何需要保持固定顺序的事件必须放在同一主题中（并且它们也必须使用相同的分区键）。最常见的是，如果事件的顺序与同一事物有关，则事件的顺序很重要。因此，根据经验，我们可以说关于同一事物的所有事件都需要在同一主题中。如果您使用事件排序方法进行数据建模，事件的排序尤为重要。这里，聚合对象的状态是通过以特定顺序重放它们来从事件日志中导出的。因此，即使可能存在许多不同的事件类型，定义聚合的所有事件也必须在同一主题中。

3. 如果一个事件涉及多个事物怎么办？例如，购买涉及产品和客户，并且从一个帐户到另一个帐户的转移涉及至少那两个帐户。我建议最初将事件记录为单个原子消息，而不是将其分成几个消息。主题，最好以完全按照您收到的方式记录事件，并尽可能采用原始形式。您可以随后使用流处理器拆分复合事件 - 但如果您过早地将其拆分，则重建原始事件要困难得多。更好的是，您可以为初始事件提供唯一ID（例如UUID）; 以后，当您将原始事件拆分为每个涉及的事物的一个事件时，您可以将该ID转发，从而使每个事件的起源都可追溯。

4. 查看消费者需要订阅的主题数量。如果几个消费者都阅读了一组特定的主题，这表明可能应该将这些主题组合在一起。如果将细粒度的主题组合成粗粒度的主题，一些消费者可能会收到他们需要忽略的不需要的事件。这不是什么大问题：消费来自Kafka的消息非常便宜，所以即使消费者最终忽略了一半的事件，这种过度消费的成本可能也不大。只有当消费者需要忽略绝大多数消息（例如99.9％是不需要的）时，我才建议从高容量流中分割低容量事件流。

5. Kafka Streams状态存储（KTable）的更改日志主题应与所有其他主题分开。在这种情况下，主题由Kafka Streams流程管理，不应与其他任何内容共享。

6. 最后，如果上述规则都没有告诉您是否将某些事件放在同一主题或不同主题中，该怎么办？然后，通过将相同类型的事件放在同一主题中，通过所有方法将它们按事件类型分组。但是，我认为这条规则是最不重要的。

模式管理

如果您使用的是数据编码（如JSON），而没有静态定义的模式，则可以轻松地将许多不同的事件类型放在同一主题中。但是，如果您使用的是基于模式的编码（如Avro），则需要更多地考虑在单个主题中处理多个事件类型。

如上所述，基于Avro的Kafka Confluent Schema Registry目前依赖于每个主题都有一个模式的假设（或者更确切地说，一个模式用于密钥，一个模式用于消息的值）。您可以注册新版本的模式，注册表会检查模式更改是向前还是向后兼容。这个设计的一个好处是，您可以让不同的生产者和消费者同时使用不同的模式版本，并且它们仍然保持彼此兼容。

更确切地说，当Confluent的Avro序列化程序在注册表中注册模式时，它会在主题名称下注册。默认情况下，该主题<topic>-key用于消息键和<topic>-value消息值。然后，模式注册表检查在特定主题下注册的所有模式的相互兼容性。

我最近对Avro序列化程序进行了修补，使兼容性检查更加灵活。该补丁添加了两个新的配置选项:(key.subject.name.strategy定义如何构造消息键的主题名称），以及value.subject.name.strategy（如何构造消息值的主题名称）。选项可以采用以下值之一：

• io.confluent.kafka.serializers.subject.TopicNameStrategy（默认值）：消息键的主题名称是<topic>-key，<topic>-value对于消息值。这意味着主题中所有消息的模式必须相互兼容。

• io.confluent.kafka.serializers.subject.RecordNameStrategy：主题名称是邮件的Avro记录类型的完全限定名称。因此，模式注册表会检查特定记录类型的兼容性，而不考虑主题。此设置允许同一主题中的任意数量的不同事件类型。

• io.confluent.kafka.serializers.subject.TopicRecordNameStrategy：主题名称是<topic>-<type>，<topic>Kafka主题名称在哪里，并且 是邮件的Avro记录类型的完全限定名称。此设置还允许同一主题中的任意数量的事件类型，并进一步将兼容性检查限制为仅当前主题。

使用此新功能，可以轻松，干净地将特定事物的所有不同事件放在同一主题中。现在，可以根据上述条件自由选择主题的粒度，而不仅限于每个主题的单个事件类型。

原文：

[1] https://www.confluent.io/blog/put-several-event-types-kafka-topic/, Martin KleppmannMartin Kleppmann