CODEWITHSUNDEEP
elasticsearchapache-kafkakafka-consumer-apiapache-kafka-connectconfluent-platform
Asim Zaidi
Asim Zaidi

Reputation: 28294

rename index in elasticsearch with kafka sink

I am using the following sink. Issue is that it sets the elasticsearch index name same as the topic. I want to have a different elasticseach index name. How can I achieve that. I am using confluent 4

{
  "name": "es-sink-mysql-foobar-02",
  "config": {
    "_comment": "-- standard converter stuff -- this can actually go in the worker config globally --",
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://localhost:8081",
    "value.converter.schema.registry.url": "http://localhost:8081",


    "_comment": "--- Elasticsearch-specific config ---",
    "_comment": "Elasticsearch server address",
    "connection.url": "http://localhost:9200",

    "_comment": "Elasticsearch mapping name. Gets created automatically if doesn't exist  ",
    "type.name": "type.name=kafka-connect",
    "index.name": "asimtest",
    "_comment": "Which topic to stream data from into Elasticsearch",
    "topics": "mysql-foobar",

    "_comment": "If the Kafka message doesn't have a key (as is the case with JDBC source)  you need to specify key.ignore=true. If you don't, you'll get an error from the Connect task: 'ConnectException: Key is used as document id and can not be null.",
    "key.ignore": "true"
  }
}

Upvotes: 1

Views: 3662

Answers (1)

Robin Moffatt
Robin Moffatt

Reputation: 32090

Use Kafka Connect's Single Message Transform (SMT) capabilities for this.

For example, to drop the mysql- prefix: 

"_comment": "Drop the mysql- prefix from the topic name and thus Elasticsearch index name",
"transforms": "dropPrefix",
"transforms.dropPrefix.type":"org.apache.kafka.connect.transforms.RegexRouter",
"transforms.dropPrefix.regex":"mysql-(.*)",
"transforms.dropPrefix.replacement":"$1"

or to drop the prefix and also route the messages to a time-based Elasticsearch index: 

 "transforms":"dropPrefix,routeTS",  
 "transforms.dropPrefix.type":"org.apache.kafka.connect.transforms.RegexRouter",  
 "transforms.dropPrefix.regex":"mysql-(.*)",  
 "transforms.dropPrefix.replacement":"$1",  
 "transforms.routeTS.type":"org.apache.kafka.connect.transforms.TimestampRouter",  
 "transforms.routeTS.topic.format":"kafka-${topic}-${timestamp}",  
 "transforms.routeTS.timestamp.format":"YYYYMM"

See https://www.confluent.io/blog/simplest-useful-kafka-connect-data-pipeline-world-thereabouts-part-3/ for more details.

Upvotes: 5

Related Questions