我只是在学习Apache Kafka。我当前的 ETL 在批处理上运行，现在我希望它在流处理上运行，以便用于报告的数据始终是最新的。 据我了解，我可以使用MongoDB连接器从mongodb捕获数据更改，然后将其发送到kafka主题。但是在我的 ETL 中，我需要在处理到 SQL 数据库后存储数据。 我如何以及在哪里处理从mongodb发送到主题的数据，然后从它创建到另一个数据库的记录？我是否可以使用 AWS lambda 函数进行处理和创建记录？但是，我如何从 kafka 调用这个函数呢？
你的问题的简短回答是Kafka Connect。较长的答案是Kafka Connect加上流处理(如Kafka Streams，ksqlDB等(。
您的管道将如下所示：
- 从带有Kafka Connect的MongoDB读取(例如，这个连接器或这个连接器(。下面是一个示例 (
- 可选(使用流处理处理数据(筛选、扩充、聚合、联接(。下面是一个使用 ksqlDB 的示例。
- 使用 Kafka Connect 和 JDBC Sink 连接器将数据接收器到数据库。下面是一个示例教程。
以下是在 ETL 中使用 Kafka 作为博客和演讲的更一般的概述。您可以在本次演讲中了解有关 Kafka Connect 的更多信息。