Kafka Connect Elasticsearch Source: fetch data from elasting indices using scroll API. Fetch only new data using a strictly incremental / temporal field. It supports dynamic schema and nested objects/ arrays.

(Good read)[https://www.confluent.io/blog/create-dynamic-kafka-connect-source-connectors/]

Download (https://github.com/DarioBalinzo/kafka-connect-elasticsearch-source/raw/master/target/kafka-connect-elastic-source-connect-0.1.jar) the jar and put into the connect classpath (e.g /usr/share/java/kafka-connect-elasticsearch ) or set plugin.path parameter appropriately.

Using kafka connect in distributed way, a sample config file to fetch metric* indices and to produce output topics with es_ prefix:

{
  "name": "elastic-source",
  "config": {
    "connector.class":"com.github.dariobalinzo.ElasticSourceConnector",
    "tasks.max": "1",
    "es.host" : "localhost",
    "es.port" : "9200",
    "es.query": "true",
    "index.prefix" : "logstash",
    "topic" : "test-topic",
    "incrementing.field.name" : "@timestamp",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://kafka-connect-cp-schema-registry.queue:8081",
    "label.key": "foo",
    "label.value": "bar"
  }
}

To start the connector we send the json config with post:

curl -X POST -H "Content-Type: application/json" --data @config.json http://localhost:8083/connectors | jq

To check the status:

curl localhost:8083/connectors/elastic-source/status | jq

es.host ElasticSearch host

• Type: string
• Importance: high
• Dependents: index.prefix

es.port ElasticSearch port

• Type: string
• Importance: high
• Dependents: index.prefix

es.user Elasticsearch username

• Type: string
• Default: null
• Importance: high

es.password Elasticsearch password

• Type: password
• Default: null
• Importance: high

connection.attempts Maximum number of attempts to retrieve a valid Elasticsearch connection.

• Type: int
• Default: 3
• Importance: low

connection.backoff.ms Backoff time in milliseconds between connection attempts.

• Type: long
• Default: 10000
• Importance: low

index.prefix Indices prefix to include in copying.

• Type: string
• Default: ""
• Importance: medium

poll.interval.ms Frequency in ms to poll for new data in each index.

• Type: int
• Default: 5000
• Importance: high

batch.max.rows Maximum number of documents to include in a single batch when polling for new data.

• Type: int
• Default: 10000
• Importance: low

topic.prefix Prefix to prepend to index names to generate the name of the Kafka topic to publish data

• Type: string
• Importance: high