如何为动态索引 Elasticsearch 配置 Kafka Connect API？

Question

大家好，当我使用 Kafka Connect API 将数据下沉到 Elasticsearch 时，我遇到了一个案例。现在我有这个问题。我的配置文件很简单

name=elasticsearch-sink
connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
tasks.max=5
topics=myTopicKafka
topic.index.map=myTopicKafka:myIndexES-1
schema.ignore=true
key.ignore=true
connection.url=http://elasticsearch:9200
type.name=kafka-connect
batch.size=200
#linger.ms=500

但在 ES 中，我使用 Curator 来滚动索引

actions:
  # 1:
  #   action: create_index
  #   description: 'Create mwe.resource.locate index'
  #   options:
  #     name: 'myIndexES-1-%Y-%m-%d-1'
  2:
    action: rollover
    description: >-
      Rollover the index associated with alias 'myIndexES', after exceeds 500MB memory or is a day old
    options:
      name: all_myIndexES
      conditions:
        max_age: 1d
        max_size: 500mb

这是每 500MB 创建一个新索引，但索引名称为 myIndexES-00002、myIndexES-00003 等。所以我在这里的问题是如何使用 Kafka Connect API 来支持这一点

Answer 1

使用Rollover API 时，您应该写入指向单个索引的别名。

这意味着在您的 elasticsearch-sink 配置中，您应该改为：

topic.index.map=myTopicKafka:myIndexES-write

在您的 Curator configuration 中，您应该有一个带有别名名称的 name 属性。

options:
  name: 'myIndexES-write'