使用 Kafka Connect 构建数据转换服务答案

【问题标题】：Build a data transformation service using Kafka Connect使用 Kafka Connect 构建数据转换服务
【发布时间】：2021-07-20 03:34:37
【问题描述】：

Kafka Streams 很好，但我必须非常手动地进行每个配置。相反，Kafka Connect 提供了它的 API 接口，这对于处理配置以及 Tasks、Worker 等非常有用......

因此，我正在考虑将 Kafka Connect 用于我的简单数据转换服务。基本上，该服务将从一个主题读取数据并将转换后的数据发送到另一个主题。为了做到这一点，我必须制作一个自定义的接收器连接器来将转换后的数据发送到 kafka 主题，但是，这些接口函数似乎在接收器连接器中不可用。如果我能做到，那就太好了，因为我可以通过 REST API 管理任务、工作人员并在分布式模式（多个实例）下运行任务。

我的想法有两个选择：

弄清楚如何将消息从 SinkConnector 发送到 kafka 主题
弄清楚如何构建一个 REST 接口 API，例如 Kafka Connect，它封装了 Kafka Streams 应用程序

有什么想法吗？

【问题讨论】：

Kafka connect 的创建是为了将 kafka 代理连接到外部系统，而不是进行流式传输。如果您认为 Kafka 流不好，您可以使用 kafka connect 将 kafka 代理连接到另一个流引擎（Spark、Flink...）。

标签： apache-kafka apache-kafka-connect

【解决方案1】：

弄清楚如何将消息从 SinkConnector 发送到 kafka 主题

接收器连接器使用来自 Kafka 主题的数据/消息。如果您想将数据发送到 Kafka 主题，您可能正在谈论源连接器。

弄清楚如何构建一个 REST 接口 API，例如 Kafka Connect，它封装了 Kafka Streams 应用程序。

使用kafka-connect-archtype，您可以使用模板来创建自己的 Kafka 连接器（源或接收器）。在您想要在连接器之后构建一些流处理管道的情况下，您主要是在谈论另一个不是 Kafka-stream 的流处理引擎的连接器。有 Kafka Spark、Kafka Flink、...

的连接器

但是如果你愿意，你可以使用kafka-connect-archtype 的模板来构建你的。使用MySourceTask List<SourceRecord> poll() 方法或MySinkTask put(Collection<SinkRecord> records) 方法将记录作为流处理。他们从 Kafka 连接扩展 org.apache.kafka.connect.[source.SourceTask|sink.SinkTask]。

【讨论】：

谢谢让我看看 kafka-connect-archtype
上周我正在实现一个 github kafka 源连接器，似乎可以使用poll() 方法。在这个例子中，我收集了一个固定大小的列表和SourceRecord (github.com/felipegutierrez/kafka-connector-github-source/blob/…) 的几个项目。在您的情况下，您将使用SinkRecord。

【解决方案2】：

一个 REST 接口 API，如 Kafka Connect，它封装了 Kafka Streams 应用程序

这正是KsqlDB允许你做的

除了使用 SQL 查询创建流和表之外，它还提供 REST API 以及可以与 Connect 端点交互（或嵌入 Connect 工作程序本身）

https://docs.ksqldb.io/en/latest/concepts/connectors/

【讨论】：