Elasticsearch #

Elasticsearch is a distributed search and analytics engine built on top of Apache Lucene. It’s not a replacement for your primary database — it’s a search layer that complements your relational database or MongoDB. The most common usage pattern: primary data is stored in PostgreSQL/MySQL, then synchronized to Elasticsearch for fast full-text search capabilities, relevance scoring, faceted search, and analytical aggregations. Elasticsearch excels at text search — it understands synonyms, stemming, spelling correction, and can return results with relevance scores. In Kotlin, you interact with Elasticsearch using the Elasticsearch Java Client (official, v8+) which communicates via REST API. This article covers the main concepts, document CRUD, various query types, aggregations, and good production patterns.

Key Elasticsearch Concepts #

Before code, understand Elasticsearch terminology:

flowchart LR
    A["Index\n(equivalent to a SQL table)"] --> B["Document\n(equivalent to a row)"]
    B --> C["Field\n(equivalent to a column)"]
    A --> D["Mapping\n(equivalent to schema/data types)"]
    A --> E["Shard\n(partition for distribution)"]
    E --> F["Replica\n(copy for reliability)"]

Elasticsearch	SQL	Description
Index	Table	A collection of similar documents
Document	Row	A unit of data in JSON format
Field	Column	A property within a document
Mapping	Schema	Definition of each field’s data type
Query	SELECT + WHERE	Document search
Aggregation	GROUP BY	Statistics and analytics

Setup and Dependencies #

// build.gradle.kts
dependencies {
    // Elasticsearch Java Client (v8+, official)
    implementation("co.elastic.clients:elasticsearch-java:8.13.0")

    // HTTP client required by the ES client
    implementation("org.apache.httpcomponents.client5:httpclient5:5.3.1")

    // Jackson for serialization (required by the ES client)
    implementation("com.fasterxml.jackson.module:jackson-module-kotlin:2.17.0")
    implementation("com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.17.0")

    // kotlinx.serialization (optional, for data models)
    implementation("org.jetbrains.kotlinx:kotlinx-serialization-json:1.6.3")
}

Creating a Client #

import co.elastic.clients.elasticsearch.ElasticsearchClient
import co.elastic.clients.json.jackson.JacksonJsonpMapper
import co.elastic.clients.transport.rest_client.RestClientTransport
import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.kotlin.registerKotlinModule
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule
import org.apache.http.HttpHost
import org.elasticsearch.client.RestClient

object ElasticsearchConnection {
    private val mapper = ObjectMapper().apply {
        registerKotlinModule()
        registerModule(JavaTimeModule())
    }

    val client: ElasticsearchClient by lazy {
        // Connection to local Elasticsearch
        val restClient = RestClient.builder(
            HttpHost("localhost", 9200, "http")
        ).build()

        val transport = RestClientTransport(restClient, JacksonJsonpMapper(mapper))
        ElasticsearchClient(transport)
    }

    // Connection with authentication (production)
    fun createClientWithAuth(
        host: String = System.getenv("ES_HOST") ?: "localhost",
        port: Int = System.getenv("ES_PORT")?.toInt() ?: 9200,
        username: String = System.getenv("ES_USER") ?: "elastic",
        password: String = System.getenv("ES_PASSWORD") ?: "changeme"
    ): ElasticsearchClient {
        val credentials = org.apache.http.auth.UsernamePasswordCredentials(username, password)
        val provider = org.apache.http.impl.client.BasicCredentialsProvider()
        provider.setCredentials(org.apache.http.auth.AuthScope.ANY, credentials)

        val restClient = RestClient.builder(HttpHost(host, port, "https"))
            .setHttpClientConfigCallback { httpClientBuilder ->
                httpClientBuilder.setDefaultCredentialsProvider(provider)
            }
            .build()

        return ElasticsearchClient(RestClientTransport(restClient, JacksonJsonpMapper(mapper)))
    }
}

Defining Mappings #

Mappings define each field’s data type. Elasticsearch can guess mappings automatically (dynamic mapping), but defining them explicitly gives more control:

import co.elastic.clients.elasticsearch.indices.CreateIndexRequest
import co.elastic.clients.elasticsearch._types.mapping.*

fun createProductIndex() {
    val client = ElasticsearchConnection.client

    // Check if the index already exists
    val alreadyExists = client.indices().exists { req ->
        req.index("produk")
    }.value()

    if (alreadyExists) {
        println("Index 'produk' already exists")
        return
    }

    // Create the index with a mapping
    client.indices().create { req ->
        req.index("produk")
            .settings { s ->
                s.numberOfShards("1")
                 .numberOfReplicas("0")  // 0 for development, 1+ for production
                 .analysis { a ->
                     a.analyzer("indonesian_analyzer") { an ->
                         an.custom { c ->
                             c.tokenizer("standard")
                              .filter(listOf("lowercase", "stop"))
                         }
                     }
                 }
            }
            .mappings { m ->
                m.properties("id", Property.of { p -> p.keyword { it } })
                 .properties("nama", Property.of { p ->
                     p.text { t ->
                         t.analyzer("indonesian_analyzer")
                          .fields("keyword", Property.of { f -> f.keyword { it.ignoreAbove(256) } })
                     }
                 })
                 .properties("deskripsi", Property.of { p ->
                     p.text { t -> t.analyzer("indonesian_analyzer") }
                 })
                 .properties("harga", Property.of { p -> p.double_ { it } })
                 .properties("stok", Property.of { p -> p.integer_ { it } })
                 .properties("kategori", Property.of { p -> p.keyword { it } })
                 .properties("tag", Property.of { p -> p.keyword { it } })
                 .properties("aktif", Property.of { p -> p.boolean_ { it } })
                 .properties("dibuat_pada", Property.of { p -> p.date_ { it.format("strict_date_time") } })
            }
    }
    println("Index 'produk' created successfully")
}

Document CRUD #

Indexing (Insert/Update) #

import co.elastic.clients.elasticsearch.core.*
import com.fasterxml.jackson.annotation.JsonProperty
import java.time.Instant

data class ProductDocument(
    val id: String,
    val nama: String,
    val deskripsi: String? = null,
    val harga: Double,
    val stok: Int = 0,
    val kategori: String? = null,
    val tag: List<String> = emptyList(),
    val aktif: Boolean = true,
    @JsonProperty("dibuat_pada") val dibuatPada: String = Instant.now().toString()
)

fun indexProduct(product: ProductDocument): String {
    val response = ElasticsearchConnection.client.index { req ->
        req.index("produk")
           .id(product.id)
           .document(product)
    }
    println("Indexed: ${response.id()} (${response.result()})")
    return response.id()
}

// Index many documents at once — the Bulk API
fun bulkIndexProducts(products: List<ProductDocument>): Int {
    val request = BulkRequest.Builder().apply {
        products.forEach { p ->
            operations { op ->
                op.index { idx ->
                    idx.index("produk").id(p.id).document(p)
                }
            }
        }
    }.build()

    val response = ElasticsearchConnection.client.bulk(request)

    val failed = response.items().count { it.error() != null }
    if (failed > 0) println("$failed items failed to index")

    return response.items().size - failed
}

Getting a Document by ID #

fun getProduct(id: String): ProductDocument? {
    val response = ElasticsearchConnection.client.get({ req ->
        req.index("produk").id(id)
    }, ProductDocument::class.java)

    return if (response.found()) response.source() else null
}

Deleting a Document #

fun deleteProduct(id: String): Boolean {
    val response = ElasticsearchConnection.client.delete { req ->
        req.index("produk").id(id)
    }
    return response.result().name == "DELETED"
}

Search #

This is Elasticsearch’s main feature. There are many query types that can be combined:

Match Query — Full-Text Search #

import co.elastic.clients.elasticsearch._types.query_dsl.*
import co.elastic.clients.elasticsearch.core.SearchResponse

fun searchProducts(keyword: String, limit: Int = 10): List<ProductDocument> {
    val response = ElasticsearchConnection.client.search({ req ->
        req.index("produk")
           .query { q ->
               // match: full-text search with analysis (stemming, lowercase, etc.)
               q.match { m ->
                   m.field("nama").query(keyword)
               }
           }
           .size(limit)
    }, ProductDocument::class.java)

    return response.hits().hits().mapNotNull { it.source() }
}

Multi-Match — Search Across Many Fields #

fun searchMultiField(keyword: String): List<ProductDocument> {
    val response = ElasticsearchConnection.client.search({ req ->
        req.index("produk")
           .query { q ->
               q.multiMatch { m ->
                   m.query(keyword)
                    .fields(listOf("nama^3", "deskripsi", "kategori"))  // ^3 = boost nama 3x
                    .type(TextQueryType.BestFields)
                    .fuzziness("AUTO")  // automatic typo tolerance
               }
           }
    }, ProductDocument::class.java)

    return response.hits().hits().mapNotNull { it.source() }
}

Bool Query — Combining Conditions #

fun searchFilteredProducts(
    keyword: String? = null,
    category: String? = null,
    minPrice: Double? = null,
    maxPrice: Double? = null,
    tag: String? = null
): List<ProductDocument> {

    val response = ElasticsearchConnection.client.search({ req ->
        req.index("produk")
           .query { q ->
               q.bool { b ->
                   // must: must match (affects the relevance score)
                   if (keyword != null) {
                       b.must { m ->
                           m.multiMatch { mm ->
                               mm.query(keyword)
                                 .fields(listOf("nama^2", "deskripsi"))
                                 .fuzziness("AUTO")
                           }
                       }
                   }

                   // filter: must match (doesn't affect the score)
                   b.filter { f -> f.term { t -> t.field("aktif").value(true) } }

                   if (category != null) {
                       b.filter { f -> f.term { t -> t.field("kategori").value(category) } }
                   }

                   if (tag != null) {
                       b.filter { f -> f.term { t -> t.field("tag").value(tag) } }
                   }

                   // Range query for price
                   if (minPrice != null || maxPrice != null) {
                       b.filter { f ->
                           f.range { r ->
                               r.field("harga").apply {
                                   if (minPrice != null) gte(co.elastic.clients.json.JsonData.of(minPrice))
                                   if (maxPrice != null) lte(co.elastic.clients.json.JsonData.of(maxPrice))
                               }
                           }
                       }
                   }

                   b
               }
           }
           .sort { s -> s.score { sc -> sc.order(co.elastic.clients.elasticsearch._types.SortOrder.Desc) } }
           .size(20)
    }, ProductDocument::class.java)

    return response.hits().hits().mapNotNull { it.source() }
}

Highlight — Mark the Matching Text #

data class SearchResult(
    val document: ProductDocument,
    val relevanceScore: Double,
    val highlight: Map<String, List<String>>
)

fun searchWithHighlight(keyword: String): List<SearchResult> {
    val response = ElasticsearchConnection.client.search({ req ->
        req.index("produk")
           .query { q ->
               q.multiMatch { m ->
                   m.query(keyword).fields(listOf("nama", "deskripsi"))
               }
           }
           .highlight { h ->
               h.preTags("<mark>").postTags("</mark>")
                .fields("nama") { it }
                .fields("deskripsi") { it.numberOfFragments(1).fragmentSize(150) }
           }
    }, ProductDocument::class.java)

    return response.hits().hits().mapNotNull { hit ->
        val source = hit.source() ?: return@mapNotNull null
        SearchResult(
            document = source,
            relevanceScore = hit.score() ?: 0.0,
            highlight = hit.highlight().mapValues { (_, v) -> v }
        )
    }
}

Aggregations — Statistics and Faceted Search #

Aggregations are very useful for filter/facet features on e-commerce search pages:

import co.elastic.clients.elasticsearch._types.aggregations.*

data class SearchFacets(
    val totalResults: Long,
    val categories: Map<String, Long>,
    val priceRanges: Map<String, Long>,
    val popularTags: Map<String, Long>
)

fun facetedSearch(keyword: String): SearchFacets {
    val response = ElasticsearchConnection.client.search({ req ->
        req.index("produk")
           .query { q ->
               q.bool { b ->
                   b.must { m -> m.match { mm -> mm.field("nama").query(keyword) } }
                   b.filter { f -> f.term { t -> t.field("aktif").value(true) } }
               }
           }
           .aggregations("per_kategori") { agg ->
               agg.terms { t -> t.field("kategori").size(20) }
           }
           .aggregations("rentang_harga") { agg ->
               agg.range { r ->
                   r.field("harga")
                    .ranges(
                        AggregationRange.of { it.key("< 1jt").to("1000000") },
                        AggregationRange.of { it.key("1-5jt").from("1000000").to("5000000") },
                        AggregationRange.of { it.key("5-15jt").from("5000000").to("15000000") },
                        AggregationRange.of { it.key("> 15jt").from("15000000") }
                    )
               }
           }
           .aggregations("tag_populer") { agg ->
               agg.terms { t -> t.field("tag").size(10) }
           }
           .size(0)  // only get aggregations, not documents
    }, ProductDocument::class.java)

    val aggs = response.aggregations()

    // Parse the category aggregation
    val categories = aggs["per_kategori"]?.sterms()?.buckets()?.array()
        ?.associate { it.key().stringValue() to it.docCount() }
        ?: emptyMap()

    // Parse the price ranges
    val priceRanges = aggs["rentang_harga"]?.range()?.buckets()?.array()
        ?.associate { (it.key() ?: "?") to it.docCount() }
        ?: emptyMap()

    // Parse the popular tags
    val popularTags = aggs["tag_populer"]?.sterms()?.buckets()?.array()
        ?.associate { it.key().stringValue() to it.docCount() }
        ?: emptyMap()

    return SearchFacets(
        totalResults = response.hits().total()?.value() ?: 0,
        categories = categories,
        priceRanges = priceRanges,
        popularTags = popularTags
    )
}

Primary Database to Elasticsearch Synchronization Patterns #

Elasticsearch isn’t the primary database — it’s a search layer. A synchronization mechanism is needed:

// Pattern: primary database (PostgreSQL) → Elasticsearch
// Data is stored in PostgreSQL, synchronized to ES for search

class ProductSyncService(
    private val dbRepo: PostgresProductRepository,  // the primary database
    private val esClient: ElasticsearchClient       // elasticsearch
) {

    // Synchronize one product after CREATE/UPDATE
    suspend fun syncProduct(id: Long) {
        val product = dbRepo.findById(id) ?: run {
            // The product was deleted from the DB — also delete from ES
            deleteFromEs(id.toString())
            return
        }

        val document = ProductDocument(
            id        = product.id.toString(),
            nama      = product.name,
            deskripsi = product.description,
            harga     = product.price.toDouble(),
            stok      = product.stock,
            kategori  = product.category,
            aktif     = product.active
        )

        esClient.index { req ->
            req.index("produk").id(document.id).document(document)
        }
        println("Synced product ${product.id} to Elasticsearch")
    }

    // Full sync — to rebuild the index from scratch
    fun fullSync(batchSize: Int = 100) {
        var page = 1
        var totalSynced = 0

        do {
            val products = dbRepo.findAll(page = page, size = batchSize)

            if (products.isEmpty()) break

            val documents = products.map { p ->
                ProductDocument(
                    id     = p.id.toString(),
                    nama   = p.name,
                    harga  = p.price.toDouble(),
                    stok   = p.stock,
                    aktif  = p.active
                )
            }

            bulkIndexProducts(documents)
            totalSynced += products.size
            println("Synced page $page: ${products.size} products")

            page++
        } while (products.size == batchSize)

        println("Full sync complete: $totalSynced products")
    }

    private fun deleteFromEs(id: String) {
        runCatching {
            esClient.delete { req -> req.index("produk").id(id) }
        }
    }
}

Production Tips #

// 1. Refresh interval — ES refreshes by default every 1 second
// For bulk indexing, disable it temporarily for better performance
fun setRefreshInterval(interval: String = "1s") {
    ElasticsearchConnection.client.indices().putSettings { req ->
        req.index("produk")
           .settings { s -> s.refreshInterval { it.time(interval) } }
    }
}

// 2. Use _source filtering to save bandwidth
fun searchWithLimitedFields(keyword: String): List<Map<String, Any?>> {
    val response = ElasticsearchConnection.client.search({ req ->
        req.index("produk")
           .query { q -> q.match { m -> m.field("nama").query(keyword) } }
           .source { s ->
               s.filter { f ->
                   f.includes(listOf("nama", "harga", "kategori"))
               }
           }
    }, Map::class.java)

    @Suppress("UNCHECKED_CAST")
    return response.hits().hits().mapNotNull { it.source() as? Map<String, Any?> }
}

// 3. Efficient pagination — use search_after for deep pagination
// Avoid from+size > 10000 (very slow)
// Use search_after with sorting by ID for page navigation

// 4. Query timeouts
fun searchWithTimeout(keyword: String): List<ProductDocument> {
    val response = ElasticsearchConnection.client.search({ req ->
        req.index("produk")
           .query { q -> q.match { m -> m.field("nama").query(keyword) } }
           .timeout("3s")  // cancel the query if it exceeds 3 seconds
    }, ProductDocument::class.java)

    if (response.timedOut()) println("Warning: query timed out!")
    return response.hits().hits().mapNotNull { it.source() }
}

Summary #

Elasticsearch isn’t the primary database — use Elasticsearch as a search layer on top of a relational database. Store primary data in PostgreSQL/MySQL, synchronize to ES for search.
Explicit mappings are better than dynamic mapping — define mappings before indexing data. Dynamic mapping can create wrong types (numbers as strings) and bloat the index.
filter vs must in bool queries — use filter for conditions that don’t affect the relevance score (active=true, category, price range). Use must for text search that affects the score.
fuzziness("AUTO") for typo tolerance — adding fuzziness to a query makes search more forgiving of user typos.
Field boost with ^N — in multi_match, use "nama^3" to give more weight to the name field over the description. Results matching the name will appear first.
Bulk API for mass indexing — for batch synchronization, always use the Bulk API instead of indexing one by one. The performance difference can be 10-100x.
Aggregations for faceted search — use terms aggregations for category filters and range aggregations for price filters. This is the foundation of filter/facet features on search pages.
Monitor with the _cat API — GET /_cat/indices?v to see all index statuses, GET /produk/_count for the document count, GET /_cluster/health for cluster health.

← Previous: MongoDB Next: Kafka →