Elasticsearch Kafka Connector：基于消息值设置索引

当然没问题！Confluent的Elasticsearch Kafka Connector刚好支持这种动态路由需求——根据每条消息里指定的字段值，把记录投递到对应的Elasticsearch索引。下面是具体的实现方法：

关键在于利用Connector的动态索引名称功能，通过占位符直接引用消息内容里的elasticsearch_index字段值，让每条消息自主决定目标索引。

必配参数详解 #

• index.name：这是实现动态路由的核心。针对你的JSON消息结构，把这个参数设置为从消息value中提取目标索引名：

  index.name=${record.value.elasticsearch_index}
  

  这样Connector会自动解析每条消息，将elasticsearch_index字段的值作为Elasticsearch的目标索引名。

• schema.ignore：设为true，因为你的消息是无Schema的原始JSON，不需要依赖Schema Registry来解析：

  schema.ignore=true
  

• value.converter：配置JSON转换器，确保能正确解析消息内容：

  value.converter=org.apache.kafka.connect.json.JsonConverter
  value.converter.schemas.enable=false
  

• key.ignore：如果消息key不需要作为Elasticsearch文档的ID或用于路由，建议设为true：

  key.ignore=true
  

完整简化配置示例 #

name=dynamic-es-sink-connector
connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
tasks.max=1
topics=your-source-kafka-topic  # 替换成你的输入Kafka主题
connection.url=http://your-es-host:9200  # 替换成你的ES地址
index.name=${record.value.elasticsearch_index}
schema.ignore=true
key.ignore=true
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
type.name=_doc

额外注意事项 #

• 若目标索引不存在，Connector默认会自动创建（auto.create.index=true是默认配置），如果需要手动管理索引，可以把这个参数设为false。
• 如果消息里的字段是嵌套结构，比如{"meta": {"elasticsearch_index": "index_1"}}，可以用点路径引用：${record.value.meta.elasticsearch_index}。
• 确保Connector拥有Elasticsearch的索引创建、写入权限，避免出现权限错误。

这样配置后，你的两条消息就会分别被投递到index_1和index_2索引中啦！

内容的提问来源于stack exchange，提问作者foxygen