Elasticsearch #

Elasticsearch adalah mesin pencari dan analitik terdistribusi yang dibangun di atas Apache Lucene. Ia bukan pengganti database utama — ia adalah lapisan pencarian yang melengkapi database relasional atau MongoDB kamu. Pola penggunaan yang paling umum: data primer disimpan di PostgreSQL/MySQL, lalu disinkronisasi ke Elasticsearch untuk kemampuan pencarian teks penuh yang cepat, relevansi scoring, faceted search, dan agregasi analitik. Elasticsearch sangat unggul dalam pencarian teks — ia mengerti sinonim, stemming, pengecekan ejaan, dan bisa mengembalikan hasil dengan skor relevansi. Di Kotlin, kamu berinteraksi dengan Elasticsearch menggunakan Elasticsearch Java Client (resmi, v8+) yang berkomunikasi via REST API. Artikel ini membahas konsep utama, CRUD dokumen, berbagai jenis query, agregasi, dan pola produksi yang baik.

Konsep Utama Elasticsearch #

Sebelum kode, pahami terminologi Elasticsearch:

flowchart LR
    A["Index\n(setara tabel di SQL)"] --> B["Document\n(setara baris)"]
    B --> C["Field\n(setara kolom)"]
    A --> D["Mapping\n(setara schema/tipe data)"]
    A --> E["Shard\n(partisi untuk distribusi)"]
    E --> F["Replica\n(salinan untuk keandalan)"]

Elasticsearch	SQL	Keterangan
Index	Tabel	Kumpulan dokumen sejenis
Document	Baris	Unit data dalam format JSON
Field	Kolom	Properti dalam dokumen
Mapping	Schema	Definisi tipe data tiap field
Query	SELECT + WHERE	Pencarian dokumen
Aggregation	GROUP BY	Statistik dan analitik

Setup dan Dependensi #

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

    // HTTP client yang dibutuhkan oleh ES client
    implementation("org.apache.httpcomponents.client5:httpclient5:5.3.1")

    // Jackson untuk serialisasi (dibutuhkan oleh 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 (opsional, untuk model data)
    implementation("org.jetbrains.kotlinx:kotlinx-serialization-json:1.6.3")
}

Membuat 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 KoneksiElasticsearch {
    private val mapper = ObjectMapper().apply {
        registerKotlinModule()
        registerModule(JavaTimeModule())
    }

    val client: ElasticsearchClient by lazy {
        // Koneksi ke Elasticsearch lokal
        val restClient = RestClient.builder(
            HttpHost("localhost", 9200, "http")
        ).build()

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

    // Koneksi dengan autentikasi (production)
    fun buatClientDenganAuth(
        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)))
    }
}

Mendefinisikan Mapping #

Mapping mendefinisikan tipe data setiap field. Elasticsearch bisa menebak mapping secara otomatis (dynamic mapping), tapi mendefinisikannya secara eksplisit memberikan kontrol lebih:

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

fun buatIndexProduk() {
    val client = KoneksiElasticsearch.client

    // Cek jika index sudah ada
    val sudahAda = client.indices().exists { req ->
        req.index("produk")
    }.value()

    if (sudahAda) {
        println("Index 'produk' sudah ada")
        return
    }

    // Buat index dengan mapping
    client.indices().create { req ->
        req.index("produk")
            .settings { s ->
                s.numberOfShards("1")
                 .numberOfReplicas("0")  // 0 untuk development, 1+ untuk 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' berhasil dibuat")
}

CRUD Dokumen #

Indexing (Insert/Update) #

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

data class DokumenProduk(
    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 indexProduk(produk: DokumenProduk): String {
    val respons = KoneksiElasticsearch.client.index { req ->
        req.index("produk")
           .id(produk.id)
           .document(produk)
    }
    println("Indexed: ${respons.id()} (${respons.result()})")
    return respons.id()
}

// Index banyak dokumen sekaligus — Bulk API
fun indexBanyakProduk(produk: List<DokumenProduk>): Int {
    val request = BulkRequest.Builder().apply {
        produk.forEach { p ->
            operations { op ->
                op.index { idx ->
                    idx.index("produk").id(p.id).document(p)
                }
            }
        }
    }.build()

    val respons = KoneksiElasticsearch.client.bulk(request)

    val gagal = respons.items().count { it.error() != null }
    if (gagal > 0) println("$gagal item gagal diindex")

    return respons.items().size - gagal
}

Mendapatkan Dokumen by ID #

fun ambilProduk(id: String): DokumenProduk? {
    val respons = KoneksiElasticsearch.client.get({ req ->
        req.index("produk").id(id)
    }, DokumenProduk::class.java)

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

Menghapus Dokumen #

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

Pencarian (Search) #

Ini adalah fitur utama Elasticsearch. Ada banyak jenis query yang bisa dikombinasikan:

Match Query — Full-Text Search #

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

fun cariProduk(kataKunci: String, limit: Int = 10): List<DokumenProduk> {
    val respons = KoneksiElasticsearch.client.search({ req ->
        req.index("produk")
           .query { q ->
               // match: full-text search dengan analisis (stemming, lowercase, dll)
               q.match { m ->
                   m.field("nama").query(kataKunci)
               }
           }
           .size(limit)
    }, DokumenProduk::class.java)

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

Multi-Match — Cari di Banyak Field #

fun cariMultiField(kataKunci: String): List<DokumenProduk> {
    val respons = KoneksiElasticsearch.client.search({ req ->
        req.index("produk")
           .query { q ->
               q.multiMatch { m ->
                   m.query(kataKunci)
                    .fields(listOf("nama^3", "deskripsi", "kategori"))  // ^3 = boost nama 3x
                    .type(TextQueryType.BestFields)
                    .fuzziness("AUTO")  // toleransi typo otomatis
               }
           }
    }, DokumenProduk::class.java)

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

Bool Query — Kombinasi Kondisi #

fun cariProdukFiltered(
    kataKunci: String? = null,
    kategori: String? = null,
    hargaMin: Double? = null,
    hargaMaks: Double? = null,
    tag: String? = null
): List<DokumenProduk> {

    val respons = KoneksiElasticsearch.client.search({ req ->
        req.index("produk")
           .query { q ->
               q.bool { b ->
                   // must: wajib cocok (mempengaruhi skor relevansi)
                   if (kataKunci != null) {
                       b.must { m ->
                           m.multiMatch { mm ->
                               mm.query(kataKunci)
                                 .fields(listOf("nama^2", "deskripsi"))
                                 .fuzziness("AUTO")
                           }
                       }
                   }

                   // filter: wajib cocok (tidak mempengaruhi skor)
                   b.filter { f -> f.term { t -> t.field("aktif").value(true) } }

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

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

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

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

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

Highlight — Tandai Teks yang Cocok #

data class HasilPencarian(
    val dokumen: DokumenProduk,
    val skorRelevansi: Double,
    val highlight: Map<String, List<String>>
)

fun cariDenganHighlight(kataKunci: String): List<HasilPencarian> {
    val respons = KoneksiElasticsearch.client.search({ req ->
        req.index("produk")
           .query { q ->
               q.multiMatch { m ->
                   m.query(kataKunci).fields(listOf("nama", "deskripsi"))
               }
           }
           .highlight { h ->
               h.preTags("<mark>").postTags("</mark>")
                .fields("nama") { it }
                .fields("deskripsi") { it.numberOfFragments(1).fragmentSize(150) }
           }
    }, DokumenProduk::class.java)

    return respons.hits().hits().mapNotNull { hit ->
        val sumber = hit.source() ?: return@mapNotNull null
        HasilPencarian(
            dokumen = sumber,
            skorRelevansi = hit.score() ?: 0.0,
            highlight = hit.highlight().mapValues { (_, v) -> v }
        )
    }
}

Aggregation — Statistik dan Faceted Search #

Aggregation sangat berguna untuk fitur filter/facet di halaman pencarian e-commerce:

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

data class FacetPencarian(
    val totalHasil: Long,
    val kategori: Map<String, Long>,
    val rentangHarga: Map<String, Long>,
    val tagPopuler: Map<String, Long>
)

fun facetedSearch(kataKunci: String): FacetPencarian {
    val respons = KoneksiElasticsearch.client.search({ req ->
        req.index("produk")
           .query { q ->
               q.bool { b ->
                   b.must { m -> m.match { mm -> mm.field("nama").query(kataKunci) } }
                   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)  // hanya ambil aggregasi, bukan dokumen
    }, DokumenProduk::class.java)

    val aggs = respons.aggregations()

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

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

    // Parse tag populer
    val tagPopuler = aggs["tag_populer"]?.sterms()?.buckets()?.array()
        ?.associate { it.key().stringValue() to it.docCount() }
        ?: emptyMap()

    return FacetPencarian(
        totalHasil = respons.hits().total()?.value() ?: 0,
        kategori = kategori,
        rentangHarga = rentangHarga,
        tagPopuler = tagPopuler
    )
}

Pola Sinkronisasi Database Utama ke Elasticsearch #

Elasticsearch bukan database utama — ia adalah lapisan pencarian. Perlu mekanisme sinkronisasi:

// Pola: database utama (PostgreSQL) → Elasticsearch
// Data disimpan di PostgreSQL, disinkronisasi ke ES untuk pencarian

class LayananSinkronisasiProduk(
    private val repoDb: ProdukRepositoryPostgres,  // database utama
    private val esClient: ElasticsearchClient       // elasticsearch
) {

    // Sinkronisasi satu produk setelah CREATE/UPDATE
    suspend fun sinkronisasiProduk(id: Long) {
        val produk = repoDb.cariById(id) ?: run {
            // Produk dihapus dari DB — hapus juga dari ES
            hapusDariEs(id.toString())
            return
        }

        val dokumen = DokumenProduk(
            id        = produk.id.toString(),
            nama      = produk.nama,
            deskripsi = produk.deskripsi,
            harga     = produk.harga.toDouble(),
            stok      = produk.stok,
            kategori  = produk.kategori,
            aktif     = produk.aktif
        )

        esClient.index { req ->
            req.index("produk").id(dokumen.id).document(dokumen)
        }
        println("Sinkronisasi produk ${produk.id} ke Elasticsearch selesai")
    }

    // Sinkronisasi penuh — untuk rebuild index dari awal
    fun sinkronisasiPenuh(batchSize: Int = 100) {
        var halaman = 1
        var totalDisinkronisasi = 0

        do {
            val produk = repoDb.cariSemua(halaman = halaman, ukuran = batchSize)

            if (produk.isEmpty()) break

            val dokumen = produk.map { p ->
                DokumenProduk(
                    id     = p.id.toString(),
                    nama   = p.nama,
                    harga  = p.harga.toDouble(),
                    stok   = p.stok,
                    aktif  = p.aktif
                )
            }

            indexBanyakProduk(dokumen)
            totalDisinkronisasi += produk.size
            println("Sinkronisasi halaman $halaman: ${produk.size} produk")

            halaman++
        } while (produk.size == batchSize)

        println("Sinkronisasi penuh selesai: $totalDisinkronisasi produk")
    }

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

Tips Produksi #

// 1. Refresh interval — ES default refresh setiap 1 detik
// Untuk indexing massal, matikan sementara untuk performa lebih baik
fun setRefreshInterval(interval: String = "1s") {
    KoneksiElasticsearch.client.indices().putSettings { req ->
        req.index("produk")
           .settings { s -> s.refreshInterval { it.time(interval) } }
    }
}

// 2. Gunakan _source filtering untuk menghemat bandwidth
fun cariDenganFieldTerbatas(kataKunci: String): List<Map<String, Any?>> {
    val respons = KoneksiElasticsearch.client.search({ req ->
        req.index("produk")
           .query { q -> q.match { m -> m.field("nama").query(kataKunci) } }
           .source { s ->
               s.filter { f ->
                   f.includes(listOf("nama", "harga", "kategori"))
               }
           }
    }, Map::class.java)

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

// 3. Pagination yang efisien — gunakan search_after untuk deep pagination
// Hindari from+size > 10000 (sangat lambat)
// Gunakan search_after dengan sort berdasarkan ID untuk navigasi halaman

// 4. Timeout query
fun cariDenganTimeout(kataKunci: String): List<DokumenProduk> {
    val respons = KoneksiElasticsearch.client.search({ req ->
        req.index("produk")
           .query { q -> q.match { m -> m.field("nama").query(kataKunci) } }
           .timeout("3s")  // batalkan query jika melebihi 3 detik
    }, DokumenProduk::class.java)

    if (respons.timedOut()) println("Peringatan: query timeout!")
    return respons.hits().hits().mapNotNull { it.source() }
}

Ringkasan #

Elasticsearch bukan database utama — gunakan Elasticsearch sebagai lapisan pencarian di atas database relasional. Simpan data primer di PostgreSQL/MySQL, sinkronisasi ke ES untuk pencarian.
Mapping eksplisit lebih baik dari dynamic mapping — definisikan mapping sebelum indexing data. Dynamic mapping bisa membuat tipe yang salah (angka sebagai string) dan membuat index membengkak.
filter vs must di bool query — gunakan filter untuk kondisi yang tidak mempengaruhi skor relevansi (aktif=true, kategori, rentang harga). Gunakan must untuk pencarian teks yang mempengaruhi skor.
fuzziness("AUTO") untuk toleransi typo — menambahkan fuzziness ke query membuat pencarian lebih forgiving terhadap kesalahan ketik pengguna.
Field boost dengan ^N — di multi_match, gunakan "nama^3" untuk memberi bobot lebih pada field nama dibanding deskripsi. Hasil yang cocok di nama akan muncul lebih dulu.
Bulk API untuk indexing massal — untuk sinkronisasi batch, selalu gunakan Bulk API daripada indexing satu per satu. Perbedaan performa bisa 10-100x.
Aggregasi untuk faceted search — gunakan terms aggregation untuk filter kategori dan range aggregation untuk filter harga. Ini adalah fondasi fitur filter/facet di halaman pencarian.
Monitoring dengan _cat API — GET /_cat/indices?v untuk melihat status semua index, GET /produk/_count untuk jumlah dokumen, GET /_cluster/health untuk kesehatan cluster.

← Sebelumnya: MongoDB Berikutnya: Kafka →