CODEWITHSUNDEEP
elasticsearchapache-kafkaapache-kafka-connect
vgrs
vgrs

Reputation: 11

unable to map topics to specified index in kafka elasticsearch connector

tried to map topic "name: localtopic" to index "name:indexoftopic" , its creating two new index in elastic search "localtopic and indexoftopic" and data of topic visible only in topic name index "localtopic", no errors in connector shown ( distributed mode )

my config is 

 "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 is created in elasticsearch but data is seen by index_name:localtopic kafkaversion:2.3 connectorversion:5 elasticsearchversion:3.2.0

even in logs INFO -- topics.regex = "", I don't know ihis option, can anyone suggest. how to use this ???

Upvotes: 1

Views: 2128

Answers (2)

Akheem Khaja
Akheem Khaja

Reputation: 76

Below worked for me, but, I was mapping only 1 topic to a different index name

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

so, with above transforms any topic such as topic1, topic1-test, topic1<anystring> will be mapped to index1

alternately, you can use the topic name in replacement as well, by changing the last 2 lines as below, this will pick 

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

basically, you can replace partial or full topic name using regex.

Upvotes: 6

OneCricketeer
OneCricketeer

Reputation: 191738

It is advised that you use RegexRouter transform instead, if you look at the config options

topic.index.map

This option is now deprecated. A future version may remove it completely. Please use single message transforms, such as RegexRouter, to map topic names to index names.

Upvotes: 2

Related Questions