自定义 Kafka Connect - ElasticSearch Sink 连接器答案

【问题标题】：Customize Kafka Connect - ElasticSearch Sink Connector自定义 Kafka Connect - ElasticSearch Sink 连接器
【发布时间】：2020-01-07 02:45:13
【问题描述】：

我的 Kafka 主题包含多种类型的消息流入并使用 Kafka Connect 写入 Elastic Search。流看起来不错，直到我必须将唯一的消息集分成唯一的索引。 IE。我必须根据字段（是 JSON 消息）获取新数据集的新索引。

如何配置/自定义 Kafka 连接来为我做同样的事情？每条消息都包含一个字段，表示消息的类型和时间戳。

示例 Json 如下所示：样本1：{"log":{"data":"information", "version":"1.1"}, "type":"xyz", "timestamp":"2019-08-28t10:07:40.370z", "value":{}}，

示例 2：{"log":{"data":"information", "version":"1.1", "value":{}}, "type":"abc", "timestamp":"2019-08-28t10:07:40.370z" }

我想自定义/配置 Kafka 连接 ES 接收器以将 Sample1 文档写入索引“xyz.20190828”，将 Sample2 文档写入索引“abc.20190828”。

我正在使用 Kafka-2.2.0 和 confluentinc-kafka-connect-elasticsearch-5.2.1 插件。

感谢您的帮助。

【问题讨论】：

很遗憾，这样的复杂操作是无法通过Kafka Connect实现的，你应该先在Kafka Streams中处理数据，然后再使用Kafka Connect推送数据到ELK

标签： elasticsearch apache-kafka apache-kafka-connect

【解决方案1】：

您可以使用write yourself 所需的自定义单消息转换 (SMT) 来执行此操作。通过根据内容更改消息的主题，您可以将其路由到不同的 Elasticsearch 索引。

目前 Apache Kafka 附带一个 SMT，它可以重命名整个主题 (RegExRouter) 或添加时间戳 (TimestampRouter)。您可能会发现这些是编写自己的有用的起点。

正如@wardzniak 在他的评论中所建议的那样，替代方法是使用流处理（例如 Kafka Streams、KSQL）对源主题进行预处理，然后再使用 Kafka Connect 将生成的单独主题发送到 Elasticsearch。

【讨论】：

我喜欢@wardzniak 的想法。想拿起来。谢谢。
我想借此机会问一下，在继续使用 KSQL/Streams 之前，我是否可以在 Elastic Search 端做一些事情来实现动态索引部分？
@srikanth 你解决了这个问题吗？我也有类似的用例。
@Neer1009 我会跳过 python 脚本的连接器。
类似于普通的 kafka 消费者使用来自 kafka 主题的消息，然后根据有效负载中的值发送到正确的弹性搜索索引？我的理解正确吗？