kafka streams - 加入分区主题
kafka streams - joining partitioned topics
我的理解是kafka streams支持分区。我想知道在连接来自两个不同主题的数据时它是如何工作的？我假设为了加入基于两个不同主题的数据，客户端应用程序必须以某种方式保证它从两个主题获得的消息共享相同的密钥。只是想知道 kafka 流是如何做到这一点的？
有几个先决条件可以进行 stream-stream 、 ktable-ktable 或 stream-ktable 连接；
- 主题需要共同划分。这意味着它们必须具有相同数量的分区。这个要求实际上很难，如果主题没有共同分区，流 API 将不允许加入，并且将在即将分配分区时在运行时抛出
TopologyBuilderException。
除此要求外，任何连接都可以工作，但为了使其正常工作，必须满足许多其他要求，例如；
- 两个主题应使用相同的密钥架构。例如，如果一个主题使用 userName 作为键，而其他 userSurname 加入操作将起作用，但很可能不会产生任何有意义的输出。
- 写入连接主题的生产者应用程序应使用相同的分区策略。这样，相同的键将最终出现在分配给要加入的相同分区中。
- 两个主题应该使用相同的消息时间戳策略（logAppendTime 或 CreateTime）。这不是每个人的要求，但如果主题使用不同的 messageTimeStampTypes，则应该考虑 windowed 连接，因为 messageTimeStamps 用于确定相关消息以连接在一起，缺少它会导致很难找到错误。
GlobalKTable 连接没有任何这些要求，并且无论分区计数、分区策略与哪个主题都适用，因为 globalKTable 的所有数据都将呈现给每个流实例。
当消息产生时，它们将根据它们的键和分区策略被发送到分区，流 API 将来自每个主题的相同主题分区分配给相同的处理器，以便来自相同主题的所有相关消息具有相同的密钥将在同一处理器中处理。对于 windowed 加入，消息时间戳被认为是为这个特定的 window 找到要加入的消息，并在加入完成后发出结果。
一个难题是确保 Kafka 流为两个主题分配相同的分区号。
为了保证这一点，它使用相同的消费者实例连接到两个主题，然后依赖范围分配器策略来获得相同的分区号。
参见 https://kafka.apache.org/24/javadoc/org/apache/kafka/clients/consumer/RangeAssignor.html
我的理解是kafka streams支持分区。我想知道在连接来自两个不同主题的数据时它是如何工作的？我假设为了加入基于两个不同主题的数据，客户端应用程序必须以某种方式保证它从两个主题获得的消息共享相同的密钥。只是想知道 kafka 流是如何做到这一点的？
有几个先决条件可以进行 stream-stream 、 ktable-ktable 或 stream-ktable 连接；
- 主题需要共同划分。这意味着它们必须具有相同数量的分区。这个要求实际上很难，如果主题没有共同分区，流 API 将不允许加入，并且将在即将分配分区时在运行时抛出
TopologyBuilderException。
除此要求外，任何连接都可以工作，但为了使其正常工作，必须满足许多其他要求，例如；
- 两个主题应使用相同的密钥架构。例如，如果一个主题使用 userName 作为键，而其他 userSurname 加入操作将起作用，但很可能不会产生任何有意义的输出。
- 写入连接主题的生产者应用程序应使用相同的分区策略。这样，相同的键将最终出现在分配给要加入的相同分区中。
- 两个主题应该使用相同的消息时间戳策略（logAppendTime 或 CreateTime）。这不是每个人的要求，但如果主题使用不同的 messageTimeStampTypes，则应该考虑 windowed 连接，因为 messageTimeStamps 用于确定相关消息以连接在一起，缺少它会导致很难找到错误。
GlobalKTable 连接没有任何这些要求，并且无论分区计数、分区策略与哪个主题都适用，因为 globalKTable 的所有数据都将呈现给每个流实例。
当消息产生时，它们将根据它们的键和分区策略被发送到分区，流 API 将来自每个主题的相同主题分区分配给相同的处理器，以便来自相同主题的所有相关消息具有相同的密钥将在同一处理器中处理。对于 windowed 加入，消息时间戳被认为是为这个特定的 window 找到要加入的消息，并在加入完成后发出结果。
一个难题是确保 Kafka 流为两个主题分配相同的分区号。 为了保证这一点，它使用相同的消费者实例连接到两个主题，然后依赖范围分配器策略来获得相同的分区号。 参见 https://kafka.apache.org/24/javadoc/org/apache/kafka/clients/consumer/RangeAssignor.html