Elasticsearch Connector

This document is based on https://docs.lenses.io/connectors/sink/elastic6.html. One use of this connector is to incorporate ElasticSearch into the existing streaming solution, where we can turn each message into ElasticSearch documents for further query.

The connectors use a specific query language called KCQL, which can be specified in a connector config for inserting into ElasticSearch. Something that we may have to keep in mind is that the connectors are used to transfer the data in its entirety between Kafka and ElasticSearch, so there is no filtering capability. This is reflected in how KCQL does not have WHERE clause to the query.

Setting up the connector is really easy, the main property that we need to set up is connect.elastic.kcql, which specify the query, and the connection properties for ElasticSearch. There will be more concrete examples when we discuss the sink connector in more detail.

For a running example, we assumed that we have the following:

1. An ElasticSearch instance.
2. A Kafka instance with topic orders-topic.

An example of CQL query and command to setup the Kafka topic as above:

Sink Connector

This connector is used to write data to an ElasticSearch instance. For full descriptions of the options, consult https://docs.lenses.io/connectors/sink/elastic6.html. This is an example of the connector config that reads data from a topic named orders-topic in Kafka and pushes it into the ElasticSearch described above.

To test this, create a kafka producer to produce some values in the form of JSON string. This is an example:

And check that the values produced are put into ElasticSearch, e.g.,  

Note that there might be differences in the exact values.

Using Other Converters

We can also use the AvroConverter if we are using Schema Registry. Add the following lines to the configuration for the connector: