自定义Kafka Connect-ElasticSearch接收器连接器

Question

我有Kafka主题，有多种类型的消息流入并使用Kafka Connect写入弹性搜索。流看起来不错，直到我不得不将唯一的消息集分离到唯一的索引中。也就是说，我必须根据字段(JSON消息)为新的数据集获取新的索引。

我如何配置/定制Kafka connect以实现同样的功能？每个消息都包含一个表示消息类型和时间戳的字段。

示例 Json 如下所示： Sample1： {“log”：{“data”：“information”， “version”：“1.1”}， “type”：“xyz”， “timestamp”：“2019-08-28t10：07：40.370z”， “value”：{}} ，

示例2：{"日志"：{"数据"："信息"，"版本"："1.1"，"值":{}}, "类型"："abc"，"时间戳"："2019-08-28t10:07:40.370z"}

我想定制/配置Kafka connect ES接收器，将Sample1文档写入索引'xyz。20190828’和Sample2文档索引为“abc.20190828”。

我用的是Kafka-2.2.0，和confluent Inc-Kafka-connect-elastic search-5 . 2 . 1插件。

感谢你的帮助。

Answer 1

您可以使用自定义的单消息转换(SMT)来实现这一点，这需要您自己编写。通过根据内容改变邮件的主题，您可以将邮件发送到不同的Elasticsearch索引。

目前，Apache Kafka附带了一个SMT，它可以重命名整个主题(< code>RegExRouter)或添加时间戳(< code>TimestampRouter)。你可能会发现这是一个有用的写作起点。

另一种选择是正如@wardzniak在他的评论中建议的那样——在使用Kafka Connect将生成的单独主题发送到Elasticsearch之前，使用流处理（例如Kafka Streams、KSQL）对源主题进行预处理。