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

Question

我目前正在尝试轻松地将消息从一个Kafka集群上的主题流式传输到另一个集群（远程）-

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

这样的事情在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。请注意，您自己的制作人必须使用同步写入！否则，在发生故障时可能会丢失数据。因此，这不是一个非常有效的解决方案。

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

编辑（回复下面的评论来自@快速洞察）

如果您的Kafka流服务中断的时间比保留时间长怎么办

这似乎是一个正交问题，可以针对任何设计提出。保留时间应根据您的最大停机时间来设置，以避免一般情况下的数据丢失。请注意，由于应用程序在源集群中读/写，并且源集群输出主题可以用很短的保留时间进行配置，因此如果应用程序宕机，不会发生什么坏事。不会处理输入主题，也不会生成新的输出数据。您可能只担心进入目标集群的复制管道中断的情况——您应该在源集群中相应地设置输出主题的保留时间，以确保不会丢失任何数据。

它还可以加倍您对Kafka的写入。

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

那看起来超级低效

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