无法将主题映射到 Kafka Elasticsearch 连接器中的指定索引

Question

试图将主题"name： localtopic"映射到索引"name：indexoftopic" ， 它在弹性搜索"localtopic"和indexoftopic"中创建两个新索引，并且主题数据仅在主题名称索引"localtopic"中可见，连接器中没有显示错误(分布式模式(

我的配置是

"config" : {
"connector.class" : "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
"tasks.max" : "1",
"topics" : "localtopic", 
"topic.index.map" : "localtopic:indexoftopic",
"connection.url" : "aws elasticsearch url",
"type.name" : "event",
"key.ignore" : "false",
"schema.ignore" : "true",
"schemas.enable" : "false",
"transforms" : "InsertKey,extractKey",
"transforms.InsertKey.type" : "org.apache.kafka.connect.transforms.ValueToKey",
"transforms.InsertKey.fields" : "event-id",
"transforms.extractKey.type" : "org.apache.kafka.connect.transforms.ExtractField$Key",
"transforms.extractKey.field" : "event-id"
}

Index Name：IndexofTopic是在Elasticsearch中创建的，但数据由index_name：localtopic看到 卡夫卡版本：2.3 连接器版本：5 弹性搜索版本：3.2.0

即使在日志信息中 - topics.regex = "，我不知道他的选择，任何人都可以建议。如何使用此???

Answer 1

下面对我有用，但是，我只将 1 个主题映射到不同的索引名称

"transforms": "addSuffix",  
"transforms.addSuffix.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.addSuffix.regex": "topic1.*",
"transforms.addSuffix.replacement": "index1",

因此，通过上述转换任何主题，例如topic1, topic1-test, topic1<anystring>将映射到index1

或者，您也可以使用主题名称替换，通过更改最后 2 行，如下所示，这将选择

"transforms.addSuffix.regex": "topic.*",
"transforms.addSuffix.replacement": "index$1",

基本上，您可以使用正则表达式替换部分或全部主题名称。

Answer 2

如果您查看配置选项，建议您改用正则表达式路由器转换

topic.index.map

此选项现已弃用。将来的版本可能会完全删除它。请使用单个消息转换(如 RegexRouter(将主题名称映射到索引名称。