使用kafka流创建基于elasticsearch事件的表

Question

是否可以使用 Kafka 流来创建从 Kafka 主题读取 JSON 的管道，然后对它们执行一些逻辑并将结果发送到另一个 Kafka 主题或其他东西？

例如，我使用来自 elasticsearch 的日志填充我的主题。使用简单的 logstash 管道非常容易。

一旦我将日志记录在 kafka 主题中，我想从日志中提取一些信息，并将它们放在一个带有 N 列的“表”中（Kafka 能做到这一点吗？），然后把表其他地方（另一个主题或数据库）。

我没有找到任何符合我标准的示例。

谢谢

Answer 1

是的，这是可能的。

在 kafka 或 kafka-streams 中没有列的概念。但是，您通常只定义您选择的普通旧 java 对象，其中包含您想要的字段（在这种情况下，字段相当于列）。 You produce the output in that format to an output topic (using an appropriately chosen serializer).最后，如果要将结果存储在关系数据库中，请将字段映射到列中，通常使用 kafka connect jdbc sink： http://docs.confluent.io/current/connect/connect-jdbc/docs/sink_connector.html