Kafka - 如何将 Rest 服务消息转换为 kafka 主题？答案

【问题标题】：Kafka - how to transform Rest service message to ingest into a kafka topic?Kafka - 如何将 Rest 服务消息转换为 kafka 主题？
【发布时间】：2019-01-25 14:09:26
【问题描述】：

我是 kafka 社区的新手，我正面临一个具有挑战性的问题。我有两个应用程序通过 Rest webservice 相互通信，正文是 json 消息。

如何使用 kafka 作为这两个应用程序之间的中间件，对应用程序的影响很小或为零？

这是我的原样场景：

我的未来情景：

我曾想过使用 STM 或一些拦截器来转换标题和正文。

这样可以吗？我知道我必须改变身体：

      {
        "first name" : "Donald" ,
        "last name" : "Trump"
      },
      {
        "first name" : "Vladimir" ,
        "last name" : "Putin"
      }

到：

{
  "records":
  [
    {
      "key": "K001",
      "value":
      {
        "first name" : "Donald" ,
        "last name" : "Trump"
      }
    },
    {
      "key": "K002",
      "value":
      {
        "first name" : "Vladimir" ,
        "last name" : "Putin"
        }
    }
  ]
}

基本上我的转换会改变消息的开头和结尾，而不是每条记录的字段。

有人做过类似的事情，或者知道我应该走哪条路吗？

【问题讨论】：

您能否提供更多背景信息来说明推动这一变化的原因？为什么要引入卡夫卡？为什么假设您仍将使用 REST？有了它，帮助会更容易:)
您好 Robin，我是解决方案架构师，我们正在更改关于集成的参考架构，我们建议从对等集成发展到中介集成。今天这两个应用是通过webservice集成的，body notation是json，我们想用kafka作为我们的中间件，不想给这些应用带来影响。这些是在将消息提取到主题之前翻译消息的主要原因。
我仍然不清楚的是，鉴于有多种语言的 Kafka 客户端库，您仍然想为此使用 REST 吗？
Robin，这两个应用程序是为了相互交流而定制的 COTS，现在我们想在这个场景中引入 kafka，对这些应用程序的影响为零
明白了，我现在明白了:)

标签： json apache-kafka confluent-platform kafka-rest

【解决方案1】：

在 Kafka 生态系统之外，您可以将 NiFi 或 Streamsets 用于您正在寻找的这些“消息转换”，两者都可以进行 HTTP 调用以及启动 HTTP 服务器本身以直接使用 API。

【讨论】：

谢谢@cricket_007，这些应用程序中的任何一个都可以解决我的问题，我什至可以将它们用作我的连接器，允许连接到数据库和restfull webservice。
没错，这些工具比 Kafka 和 Kafka Connect 提供的更灵活

【解决方案2】：

您可能需要一个 Kafka Streams 应用程序来执行此操作。它将订阅应用程序 A 通过 REST 填充的源主题，根据需要转换消息并将其写入新主题，应用程序 B 可以从中使用它。

Kafka Streams 是 Apache Kafka 的一部分，是一个 Java 库，您可以使用它构建流处理应用程序。还有 KSQL，它是 Kafka Streams 之上的一种抽象，采用类似 SQL 的语言，但我不确定它是否支持您在此处查看的转换类型。

由于您没有使用 Kafka Connect，因此单消息转换 (SMT) 不适用于此处。

【讨论】：

Robin，让我看看我是否理解正确，根据你的建议，我应该创建一个 kafka 流来读取一个主题的信息并将其写入另一个主题，是吗？我无法将消息写入主题，因为应用程序 A（生产者）调用 Web 服务，该服务具有标头：“Content-Type：application/json”，而不是标头：“Content-Type：application/vnd .kafka.json.v2+json" 和带有值集合的消息，没有消息的开头：{ "records": [，没有集合："key": "K00x", "value": { data }，在每组数据之前。