elasticsearch - 如何根据Kafka消息中的消息字段在elasticsearch上创建索引？

Question

我们有我们的中央日志平台，它使用 Fluentd 将日志从 1000 多个实例发送到 Elasticsearch 集群。并且由于 ES 集群上的大量并行写入，我们开始看到频繁的性能停机。现在我们的用例已更改为日志流。在那里，我们决定一起去Fluentd ==> Kafka ==> Elasticsearch。在我们当前的架构中，我们根据 Project 和 Env 在 Elasticsearch 上创建单独的索引。Project-Env-YYYY.MM.DD 我们以希望检查的格式创建滚动索引，以检查我们是否将所有实例的日志推送到单个主题，例如central-logs 日志消息具有以下字段：

{
project: 'abc'
env: 'staging'
...
...
..
..
}

那么我们如何使用 kafka connect 在 Elasticsearch 上创建索引，格式为Project-Env-YYYY.MM.DD

标签： elasticsearchapache-kafkaapache-kafka-connect