Elasticsearch Sink Connector

Summary

Elasticsearch is a distributed search and analytics engine. The Elasticsearch Sink Connector streams data from Kafka into Elasticsearch indices, making your data immediately searchable.

WHAT IS THE ELASTICSEARCH SINK CONNECTOR?

The Elasticsearch Sink Connector consumes messages from Kafka topics and indexes them into Elasticsearch. It handles document creation, updates, and deletions, with configurable indexing strategies.

The connector uses Elasticsearch's Bulk API to index documents efficiently. Messages from Kafka become documents in Elasticsearch, instantly searchable and ready for analytics.

USE CASES

Common use cases for the Elasticsearch Sink Connector include:

Log Aggregation: Stream application logs from Kafka to Elasticsearch for centralized search and analysis. Combine with Kibana for visualization.

Real-Time Search: Index product catalogs, user profiles, or content in real-time, keeping search results always up-to-date.

Metrics and Monitoring: Index system metrics and events for operational dashboards and alerting.

Full-Text Search: Enable powerful search capabilities over your Kafka data with Elasticsearch's query DSL.

CONNECTOR CONFIGURATION

The following shows a typical Elasticsearch Sink configuration.

1{
2  "name": "elasticsearch-sink",
3  "config": {
4    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
5    "tasks.max": "1",
6    "topics": "logs,metrics,events",
7    "connection.url": "http://localhost:9200",
8    "type.name": "_doc",
9    "key.ignore": "false",
10    "schema.ignore": "false",
11    "batch.size": "2000",
12    "max.buffered.records": "20000",
13    "flush.timeout.ms": "30000",
14    "max.in.flight.requests": "5",
15    "max.retries": "5",
16    "retry.backoff.ms": "100"
17  }
18}

Key properties:

connection.url: Elasticsearch cluster endpoint.

type.name: Document type (use _doc for Elasticsearch 7+).

topics: Comma-separated list of topics to consume.

batch.size: Number of documents to batch before sending to Elasticsearch.

key.ignore: If true, use Kafka offset as document ID instead of message key.

flush.timeout.ms: Maximum time to wait before flushing a batch.

The connector automatically creates indices based on topic names.

DOCUMENT ID STRATEGIES

The connector offers different strategies for setting Elasticsearch document IDs.

Using the Kafka message key as the document ID enables idempotent writes - re-consuming the same message updates the existing document instead of creating duplicates.

INDEXING FLOW

Let's see how documents flow from Kafka to Elasticsearch.

The connector batches messages and uses Elasticsearch's Bulk API for efficient indexing. This dramatically improves throughput compared to indexing documents one at a time.

INDEX NAMING STRATEGIES

The connector supports flexible index naming.

Default: Index name equals topic name (topic logs → index logs)

With Prefix: Add a prefix to index names

1"index.prefix": "kafka-"

Topic logs → index kafka-logs

Time-Based: Create time-based indices for better management

1"behavior.on.null.values": "ignore",
2"topic.index.map": "logs:logs-<yyyy-MM-dd>"

Topic logs → index logs-2024-01-15

Time-based indices are common for log data, making it easy to delete old data and optimize search performance.

SCHEMA AND DATA TYPES

The connector automatically maps Kafka schema types to Elasticsearch data types.

Kafka Type → Elasticsearch Type

• INT32, INT64 → long
• FLOAT32, FLOAT64 → double
• STRING → text with keyword field
• BOOLEAN → boolean
• BYTES → binary
• STRUCT → object
• ARRAY → array

The connector handles schema evolution by updating Elasticsearch mappings when new fields appear in Kafka messages.

HANDLING DELETES

The connector supports document deletion using tombstone records.

Configuration

1"behavior.on.null.values": "delete"

When the connector consumes a message with a null value (tombstone), it deletes the corresponding document from Elasticsearch. This enables CDC-style workflows where deletions in the source are reflected in Elasticsearch.

PERFORMANCE TUNING

Key settings for optimizing throughput:

batch.size: Increase for higher throughput (default: 2000)

max.buffered.records: Maximum records to buffer (default: 20000)

flush.timeout.ms: Maximum wait time before flushing

max.in.flight.requests: Concurrent requests to Elasticsearch

Balance throughput against memory usage and indexing latency.

Let's recap what we learned about the Elasticsearch Sink Connector:

Key ideas

The Elasticsearch Sink Connector streams Kafka messages to Elasticsearch indices for real-time search and analytics.

Common use cases include log aggregation, real-time search, metrics monitoring, and full-text search.

The connector uses Elasticsearch's Bulk API for efficient batch indexing.

Document IDs can come from Kafka keys, offsets, or custom fields, enabling idempotent writes.

Flexible index naming supports time-based indices for better data management.

Schema mapping is automatic, with support for schema evolution.

Tombstone records enable document deletion for CDC workflows.

Performance tuning via batch size, buffering, and concurrent requests optimizes throughput.