使用 Spring Cloud Dataflow 从 Kafka 到 MongoDB答案

【问题标题】：Kafka to MongoDB using Spring Cloud Dataflow使用 Spring Cloud Dataflow 从 Kafka 到 MongoDB
【发布时间】：2020-01-30 14:35:34
【问题描述】：

我正在做一个项目，我必须处理来自 Kafka 集群的数据，对其进行处理并将其发送到 MongoDB。该应用程序应该可以部署在 Pivotal Cloud 基础上。在网上做了一些研究后，我发现工具包 Spring-Cloud-Dataflow 很有趣，因为它可以部署在 PCF 中。我想知道我们如何使用它来创建我们的实时流管道。目前，我正在考虑使用 Kafka Streams 和 Spring Cloud Stream 来处理和转换主题流，但我不知道如何将它集成到 SCDF 中，也不知道如何将这些流发送到 MongoDB。如果我的问题不清楚，我很抱歉，我对这些框架完全陌生。

提前致谢

【问题讨论】：

Mongo 开发了自己的 Kafka 连接器。为什么不使用它？也可以部署到 PCF 上，几乎不需要代码
感谢您的回复。正如我所说，在写入 MongoDB 之前需要进行一些处理和转换:)
这可以通过 Spring-Kafka / Cloud Streams 中的 Kafka Streams API 来完成。从 Kafka 读取，写回单独的主题，然后将处理后的主题数据拉出到 Mongo

标签： mongodb apache-kafka spring-cloud-stream cloud-foundry spring-cloud-dataflow

【解决方案1】：

您可以使用 SCDF 中的 named-destination 支持直接使用来自 Kafka 或任何其他支持 message broker implementations 的 Spring Cloud Stream 的事件。

现在，对于写入部分，您可以使用我们构建、维护和发布的开箱即用的 MongoDB-sink 应用程序。

如果您必须在写入 MongoDB 之前进行一些处理，您可以使用所需的 binder 实现创建自定义 Spring Cloud Stream 应用程序 [参见：dev-guide/docs]。

总而言之，如果我们假设您有来自名为 Customers 的 Kafka 主题的事件，并且自定义处理器对每个接收到的有效负载进行一些转换（假设处理器的名称为 CustomerTransformer )，最后是 MongoDB 的写作部分。

以下是根据 SCDF 的仪表板设计的流式数据管道用例：

【讨论】：

感谢您的回复。我仍然对您的解决方案有一些疑问。 1- 使用开箱即用的 MongoDB-sink 应用程序和已经内置在 SCDF 中的应用程序有什么区别？ 2- 流必须可部署到 Pivotal Cloud Foundary 中。我们的应用程序是否有任何更改？ 3- 像 CustomerTransformer 这样的自定义处理器可以将两个主题作为源并使用使用 Spring Cloud Stream 构建的 kafka 流在它们之间进行连接吗？提前谢谢你:)
1) 应用在 SCDF 中注册。 MongoDB sink 是我们构建的一个应用程序，它可以在 Stream 中按原样使用。没有其他区别。 2) PCF 或 K8s 不需要特殊处理——应用程序的工作原理完全相同。 3) 请阅读 Spring Cloud Stream 文档，是的，多个输入和输出通道是可能的。 Kafka Streams 也可以。见Spring Cloud Stream samples。
非常感谢您的快速回复。与 SCS 有很多关系，我会阅读文档 :)
我开始学习 SCDF，发现有很多版本：Hoxton、Greenwich、Finchley、Edgware、Dalston。你会推荐哪一个？
您可能需要选择最新的 GA 版本的 Spring Cloud Stream。版本矩阵列在website。