将消息从一个 Kafka 集群流式传输到另一个

Question

我目前正在尝试轻松地将消息从一个 Kafka 集群上的主题流式传输到另一个（远程 -> 本地集群）。
我们的想法是立即使用 Kafka-Streams，这样我们就不需要在本地集群上复制实际消息，而只需将 Kafka-Streams 处理的“结果”获取到我们的 Kafka-Topics 中。

假设 WordCount 演示在我自己的另一台 PC 上的一个 Kafka-Instance 上。我还在本地机器上运行了一个 Kafka-Instance。
现在我想让 WordCount 演示在包含应计算单词的句子的主题（“远程”）上运行。
然而，计数应该写入我本地系统上的主题而不是“远程”主题。

使用 Kafka-Streams API 可以实现类似的操作吗？
例如。

val builder: KStreamBuilder = new KStreamBuilder(remote-streamConfig, local-streamconfig)
val textLines: KStream[String, String] = builder.stream("remote-input-topic", 
remote-streamConfig)
val wordCounts: KTable[String, Long] = textLines
    .flatMapValues(textLine => textLine.toLowerCase.split("\\W+").toIterable.asJava)
    .groupBy((_, word) => word)
    .count("word-counts")

wordCounts.to(stringSerde, longSerde, "local-output-topic", local-streamconfig)

val streams: KafkaStreams = new KafkaStreams(builder)
streams.start()

非常感谢
- 蒂姆

Answer 1

Kafka Streams 仅针对单个集群构建。

一种解决方法是使用foreach() 或类似名称并实例化您自己的KafkaProducer 以写入目标集群。请注意，您自己的制作人必须使用同步写入！否则，如果发生故障，您可能会丢失数据。因此，这不是一个非常高效的解决方案。

最好只将结果写入源集群并将数据复制到目标集群。请注意，您很可能可以在源集群中使用更短的输出主题保留期，因为实际数据在目标集群中的保留时间更长。这允许您限制源集群上所需的存储空间。

编辑（从@quickinsights 回复下面的评论）

如果您的 Kafka 流服务停机时间超过保留时间，该怎么办

这似乎是一个正交问题，可以针对任何设计提出。应根据您的最大停机时间设置保留时间，以避免一般数据丢失。请注意，由于应用程序从源集群读取/写入，并且源集群输出主题可能配置了较小的保留时间，因此如果应用程序出现故障，也不会发生任何不良情况。输入主题不会被处理，也不会产生新的输出数据。您可能只担心到目标集群的复制管道出现故障的情况——您应该相应地设置输出主题在源集群中的保留时间，以确保不会丢失任何数据。

它还会将您写回 Kafka 的数据翻倍。

是的。它还增加了磁盘上的存储空间。这是应用程序弹性和运行时性能与集群负载之间的权衡（一如既往）。你的选择。如上所述，我个人建议使用更具弹性的选项。扩展 Kafka 集群比处理应用程序代码中的所有弹性边缘情况更容易。

这似乎超级低效

这是个人判断。这是一种权衡，没有客观的对错。