Elasticsearch #

Elasticsearch is a distributed search and analytics engine built on Apache Lucene, designed for full-text search, logging, monitoring, and real-time data analytics at scale. Unlike traditional databases optimized for storing and retrieving exact data, Elasticsearch is optimized for searching, analyzing, and visualizing data in milliseconds — even on terabyte-sized datasets. Python interacts with Elasticsearch through the official elasticsearch-py library, and since version 8.x its API changed significantly: the body= parameter was removed, authentication became mandatory, and TLS is enabled by default.

Installation #

pip install elasticsearch

Make sure the library version matches your Elasticsearch server version. Elasticsearch 8.x is not compatible with the 7.x library and vice versa. Use pip install "elasticsearch>=8,<9" to ensure the right version.

Elasticsearch Basic Concepts #

Before writing code, it’s important to understand Elasticsearch terminology:

Elasticsearch	Relational Database
Index	Table
Document	Row
Field	Column
Mapping	Schema / DDL
Shard	Horizontal partition

Elasticsearch-specific concepts:

Inverted Index — the core data structure for full-text search, mapping keywords to lists of documents.
Analyzer — a text-processing pipeline made up of a tokenizer (word splitter) and filters (word cleaners/filters).
Relevance Score — the numeric score (_score) indicating how well a document matches the search query.
Aggregation — analytic features for grouping data (buckets) and computing metrics (e.g., sum, avg, min, max).

Creating a Connection #

Elasticsearch 8.x enables TLS and authentication by default. The connection method differs depending on whether you use local Elasticsearch, a self-managed cluster, or Elastic Cloud.

from elasticsearch import Elasticsearch
import os

# ANTI-PATTERN: old connection style (ES 7.x format, insecure)
es = Elasticsearch([{"host": "localhost", "port": 9200}])  # ✗ -- deprecated

# CORRECT: ES 8.x local connection with an API key or basic auth
es = Elasticsearch(
    hosts=os.getenv("ES_HOST", "https://localhost:9200"),
    api_key=os.getenv("ES_API_KEY"),          # recommended
    verify_certs=False,                        # dev/self-signed cert only
    ssl_show_warn=False
)

# Or with a username/password
es = Elasticsearch(
    hosts=os.getenv("ES_HOST", "https://localhost:9200"),
    basic_auth=(
        os.getenv("ES_USER", "elastic"),
        os.getenv("ES_PASSWORD")
    ),
    verify_certs=False
)

# Connecting to Elastic Cloud (production)
es = Elasticsearch(
    cloud_id=os.getenv("ES_CLOUD_ID"),
    api_key=os.getenv("ES_API_KEY")
)

# Test the connection
info = es.info()
print(f"Elasticsearch {info['version']['number']} — {info['cluster_name']}")

API keys are the recommended authentication method for production. Create an API key in Kibana → Security → API Keys, or via the API es.security.create_api_key(). API keys can have restricted permissions (read-only, specific indexes), which is safer than a username/password.

Creating an Index and Mapping #

A mapping defines the type of every field in an index — similar to a schema in a relational database. Elasticsearch can auto-detect types (dynamic mapping), but an explicit mapping is safer for production.

INDEX_PRODUK = "produk"

def buat_index_produk(es: Elasticsearch) -> None:
    # Delete the index if it already exists (for dev/testing)
    if es.indices.exists(index=INDEX_PRODUK):
        es.indices.delete(index=INDEX_PRODUK)

    es.indices.create(
        index=INDEX_PRODUK,
        mappings={
            "properties": {
                "nama": {
                    "type": "text",                  # analyzed for full-text search
                    "analyzer": "indonesian",         # Indonesian analyzer
                    "fields": {
                        "keyword": {"type": "keyword"}  # for exact match and sorting
                    }
                },
                "deskripsi": {
                    "type": "text",
                    "analyzer": "indonesian"
                },
                "harga": {
                    "type": "scaled_float",
                    "scaling_factor": 100            # stored as integer * 100
                },
                "stok":      {"type": "integer"},
                "kategori":  {"type": "keyword"},    # exact match, not analyzed
                "tag":       {"type": "keyword"},    # array of keywords
                "aktif":     {"type": "boolean"},
                "rating":    {"type": "float"},
                "dibuat_pada": {
                    "type":   "date",
                    "format": "strict_date_optional_time"
                },
                "spesifikasi": {"type": "object"},   # embedded object, free fields
                "lokasi": {
                    "type": "geo_point"              # GPS coordinates (optional)
                }
            }
        },
        settings={
            "number_of_shards":   1,     # shard for dev (production: adjust to data size)
            "number_of_replicas": 0,     # replicas for dev (production: at least 1)
            "analysis": {
                "analyzer": {
                    "indonesian": {
                        "type":      "custom",
                        "tokenizer": "standard",
                        "filter":    ["lowercase", "asciifolding"]
                    }
                }
            }
        }
    )
    print(f"Index '{INDEX_PRODUK}' created successfully.")

buat_index_produk(es)

Indexing Documents #

from datetime import datetime, timezone

# Index one document -- ES auto-generates the ID
def tambah_produk(es: Elasticsearch, produk: dict) -> str:
    produk["dibuat_pada"] = datetime.now(timezone.utc).isoformat()
    
    hasil = es.index(index=INDEX_PRODUK, document=produk)
    return hasil["_id"]

# Index with an explicit ID
def tambah_produk_dengan_id(es: Elasticsearch, produk_id: str, produk: dict) -> str:
    produk["dibuat_pada"] = datetime.now(timezone.utc).isoformat()
    
    hasil = es.index(index=INDEX_PRODUK, id=produk_id, document=produk)
    return hasil["_id"]

# Fetch a document by ID
def ambil_produk(es: Elasticsearch, produk_id: str) -> dict | None:
    try:
        hasil = es.get(index=INDEX_PRODUK, id=produk_id)
        return {"id": hasil["_id"], **hasil["_source"]}
    except Exception:
        return None

# Usage example
id1 = tambah_produk(es, {
    "nama":      "Laptop Gaming ASUS ROG Strix G15",
    "deskripsi": "High-performance gaming laptop with an AMD Ryzen 9 processor",
    "harga":     18500000,
    "stok":      10,
    "kategori":  "Elektronik",
    "tag":       ["laptop", "gaming", "asus", "rog"],
    "aktif":     True,
    "rating":    4.7
})
print(f"Document added, ID: {id1}")

# Refresh so the document is immediately searchable
es.indices.refresh(index=INDEX_PRODUK)

Bulk Indexing #

To index many documents at once, use helpers.bulk(), far more efficient than a loop of individual index() calls.

from elasticsearch import helpers

def bulk_index_produk(es: Elasticsearch, daftar_produk: list[dict]) -> tuple[int, list]:
    def generate_actions():
        for produk in daftar_produk:
            yield {
                "_index":  INDEX_PRODUK,
                "_id":     produk.get("id"),       # None = auto-generate
                "_source": {
                    **produk,
                    "dibuat_pada": datetime.now(timezone.utc).isoformat()
                }
            }

    sukses, gagal = helpers.bulk(
        es,
        generate_actions(),
        chunk_size=500,          # send per 500 documents
        request_timeout=30
    )
    return sukses, gagal

data_produk = [
    {"nama": "iPhone 15 Pro", "harga": 20000000, "kategori": "Elektronik", "stok": 5, "rating": 4.9, "aktif": True},
    {"nama": "Samsung Galaxy S24", "harga": 15000000, "kategori": "Elektronik", "stok": 8, "rating": 4.6, "aktif": True},
    {"nama": "Sepatu Lari Nike Air Max", "harga": 1800000, "kategori": "Olahraga", "stok": 20, "rating": 4.5, "aktif": True},
    {"nama": "Kopi Arabika Gayo 500gr", "harga": 85000, "kategori": "Makanan", "stok": 100, "rating": 4.8, "aktif": True},
]

sukses, gagal = bulk_index_produk(es, data_produk)
print(f"Success: {sukses}, Failed: {len(gagal)}")
es.indices.refresh(index=INDEX_PRODUK)

Full-Text Search #

This is Elasticsearch’s main advantage over regular databases — the ability to search text naturally, tolerant of word variations.

Match Query #

# match -- full-text search on a single field
def cari_produk_sederhana(es: Elasticsearch, kata_kunci: str) -> list[dict]:
    hasil = es.search(
        index=INDEX_PRODUK,
        query={
            "match": {
                "nama": {
                    "query":    kata_kunci,
                    "operator": "and"     # all words must be present (default: "or")
                }
            }
        }
    )
    return [
        {"id": h["_id"], "score": h["_score"], **h["_source"]}
        for h in hasil["hits"]["hits"]
    ]

# multi_match -- search several fields at once
def cari_multi_field(es: Elasticsearch, kata_kunci: str) -> list[dict]:
    hasil = es.search(
        index=INDEX_PRODUK,
        query={
            "multi_match": {
                "query":  kata_kunci,
                "fields": ["nama^3", "deskripsi", "tag"],  # ^3 = boost the nama field 3x
                "type":   "best_fields"
            }
        }
    )
    return [
        {"id": h["_id"], "score": round(h["_score"], 2), **h["_source"]}
        for h in hasil["hits"]["hits"]
    ]

# match_phrase -- search for an exact phrase
def cari_frasa(es: Elasticsearch, frasa: str) -> list[dict]:
    hasil = es.search(
        index=INDEX_PRODUK,
        query={"match_phrase": {"nama": frasa}}
    )
    return [{"id": h["_id"], **h["_source"]} for h in hasil["hits"]["hits"]]

Bool Query — Combining Conditions #

The bool query is the most flexible way to combine various search conditions.

def cari_produk_lanjutan(
    es:          Elasticsearch,
    kata_kunci:  str  = None,
    kategori:    str  = None,
    harga_min:   float = None,
    harga_max:   float = None,
    rating_min:  float = None,
    hanya_aktif: bool  = True,
) -> list[dict]:

    must    = []   # MUST match, affects the score
    filter_ = []   # MUST match, does NOT affect the score (more efficient)
    should  = []   # SHOULD match (boosts the score if matched)

    if kata_kunci:
        must.append({
            "multi_match": {
                "query":  kata_kunci,
                "fields": ["nama^3", "deskripsi"],
                "type":   "best_fields"
            }
        })

    if kategori:
        filter_.append({"term": {"kategori": kategori}})

    if hanya_aktif:
        filter_.append({"term": {"aktif": True}})

    range_harga = {}
    if harga_min is not None:
        range_harga["gte"] = harga_min
    if harga_max is not None:
        range_harga["lte"] = harga_max
    if range_harga:
        filter_.append({"range": {"harga": range_harga}})

    if rating_min is not None:
        filter_.append({"range": {"rating": {"gte": rating_min}}})

    query = {"bool": {}}
    if must:
        query["bool"]["must"] = must
    if filter_:
        query["bool"]["filter"] = filter_
    if should:
        query["bool"]["should"] = should
    if not must and not filter_:
        query = {"match_all": {}}

    hasil = es.search(
        index=INDEX_PRODUK,
        query=query,
        sort=[
            {"_score":  {"order": "desc"}},
            {"rating":  {"order": "desc"}},
            {"harga":   {"order": "asc"}}
        ],
        size=20
    )

    return [
        {
            "id":    h["_id"],
            "score": round(h["_score"] or 0, 2),
            **h["_source"]
        }
        for h in hasil["hits"]["hits"]
    ]

# Usage example
hasil = cari_produk_lanjutan(
    es,
    kata_kunci="laptop gaming",
    kategori="Elektronik",
    harga_max=20000000,
    rating_min=4.5
)
for p in hasil:
    print(f"[{p['score']}] {p['nama']} — Rp{p['harga']:,.0f}")

Pagination and Highlighting #

def cari_dengan_paginasi(
    es:         Elasticsearch,
    kata_kunci: str,
    halaman:    int = 1,
    per_halaman: int = 10
) -> dict:
    offset = (halaman - 1) * per_halaman

    hasil = es.search(
        index=INDEX_PRODUK,
        query={
            "multi_match": {
                "query":  kata_kunci,
                "fields": ["nama^2", "deskripsi"]
            }
        },
        highlight={
            "fields": {
                "nama":      {"number_of_fragments": 0},   # show the whole field
                "deskripsi": {"fragment_size": 150, "number_of_fragments": 2}
            },
            "pre_tags":  ["<mark>"],    # opening HTML highlight tag
            "post_tags": ["</mark>"]    # closing HTML highlight tag
        },
        from_=offset,
        size=per_halaman,
        track_total_hits=True          # count the total matching documents
    )

    total = hasil["hits"]["total"]["value"]
    hits  = hasil["hits"]["hits"]

    return {
        "total":        total,
        "halaman":      halaman,
        "per_halaman":  per_halaman,
        "total_halaman": -(-total // per_halaman),   # ceiling division
        "hasil": [
            {
                "id":        h["_id"],
                "score":     round(h["_score"], 2),
                **h["_source"],
                "highlight": h.get("highlight", {})
            }
            for h in hits
        ]
    }

response = cari_dengan_paginasi(es, "laptop gaming", halaman=1)
print(f"Total: {response['total']} results")
for item in response["hasil"]:
    print(f"- {item['nama']}")
    if "nama" in item["highlight"]:
        print(f"  → {item['highlight']['nama'][0]}")

Aggregations #

Elasticsearch aggregations enable real-time analytics — computing distributions, statistics, and trends from already-indexed data.

def analitik_produk(es: Elasticsearch) -> dict:
    hasil = es.search(
        index=INDEX_PRODUK,
        query={"term": {"aktif": True}},
        size=0,    # 0 = return only aggregations, not documents
        aggs={
            # Bucket aggregation: group by category
            "per_kategori": {
                "terms": {
                    "field": "kategori",
                    "size":  20
                },
                "aggs": {
                    # Nested metric aggregation within each bucket
                    "rata_harga":  {"avg":   {"field": "harga"}},
                    "harga_min":   {"min":   {"field": "harga"}},
                    "harga_max":   {"max":   {"field": "harga"}},
                    "rata_rating": {"avg":   {"field": "rating"}},
                    "total_stok":  {"sum":   {"field": "stok"}},
                }
            },

            # Overall statistics
            "statistik_harga": {
                "extended_stats": {"field": "harga"}
            },

            # Price histogram -- distribution within a certain range
            "distribusi_harga": {
                "histogram": {
                    "field":    "harga",
                    "interval": 5000000    # per 5 million
                }
            },

            # Range aggregation -- buckets based on custom ranges
            "segmen_harga": {
                "range": {
                    "field": "harga",
                    "ranges": [
                        {"key": "Budget",    "to":   1000000},
                        {"key": "Menengah",  "from": 1000000, "to": 5000000},
                        {"key": "Premium",   "from": 5000000}
                    ]
                }
            }
        }
    )

    aggs = hasil["aggregations"]

    # Show per category
    print("=== Per Category ===")
    for bucket in aggs["per_kategori"]["buckets"]:
        print(f"{bucket['key']}: {bucket['doc_count']} products, "
              f"average Rp{bucket['rata_harga']['value']:,.0f}, "
              f"rating {bucket['rata_rating']['value']:.1f}")

    # Price distribution
    print("\n=== Price Segments ===")
    for bucket in aggs["segmen_harga"]["buckets"]:
        print(f"{bucket['key']}: {bucket['doc_count']} products")

    return aggs

Updating and Deleting Documents #

# Update some fields (partial update)
def update_produk(es: Elasticsearch, produk_id: str, perubahan: dict) -> bool:
    try:
        es.update(
            index=INDEX_PRODUK,
            id=produk_id,
            doc=perubahan    # only the included fields are changed
        )
        return True
    except Exception as e:
        print(f"Update failed: {e}")
        return False

# Update using a script (for atomic operations)
def increment_stok(es: Elasticsearch, produk_id: str, jumlah: int) -> bool:
    try:
        es.update(
            index=INDEX_PRODUK,
            id=produk_id,
            script={
                "source": "ctx._source.stok += params.jumlah",
                "params": {"jumlah": jumlah}
            }
        )
        return True
    except Exception:
        return False

# Update by query -- update many documents at once
def nonaktifkan_kategori(es: Elasticsearch, kategori: str) -> int:
    hasil = es.update_by_query(
        index=INDEX_PRODUK,
        query={"term": {"kategori": kategori}},
        script={"source": "ctx._source.aktif = false"}
    )
    return hasil["updated"]

# Delete a document
def hapus_produk(es: Elasticsearch, produk_id: str) -> bool:
    try:
        es.delete(index=INDEX_PRODUK, id=produk_id)
        return True
    except Exception:
        return False

# Delete by query
def hapus_produk_tidak_aktif(es: Elasticsearch) -> int:
    hasil = es.delete_by_query(
        index=INDEX_PRODUK,
        query={"term": {"aktif": False}}
    )
    return hasil["deleted"]

Error Handling #

from elasticsearch import (
    Elasticsearch,
    NotFoundError,
    ConflictError,
    ConnectionError,
    TransportError
)

def ambil_produk_aman(es: Elasticsearch, produk_id: str) -> dict | None:
    try:
        hasil = es.get(index=INDEX_PRODUK, id=produk_id)
        return {"id": hasil["_id"], **hasil["_source"]}
    except NotFoundError:
        return None   # document not found
    except ConnectionError:
        print("Can't connect to Elasticsearch.")
        return None
    except TransportError as e:
        print(f"Transport error [{e.status_code}]: {e.error}")
        return None

When to Use Elasticsearch #

Use Elasticsearch for:
  ✓ Full-text search with relevance ranking (e-commerce, blogs, documentation)
  ✓ Log aggregation and real-time analytics (ELK Stack)
  ✓ Auto-complete and search-as-you-type
  ✓ Faceted search (filter by category, price, rating)
  ✓ Big-data analytics with complex aggregations

Don't use it as a primary database because:
  ✗ It doesn't support ACID transactions
  ✗ JOIN operations between indexes aren't natively supported
  ✗ Document updates are slower than in relational databases
  ✗ Storage is larger because of the inverted index

Common pattern: primary database (PostgreSQL/MySQL) + Elasticsearch as the search layer

Summary #

Elasticsearch 8.x requires authentication — use an API key (recommended) or basic auth; don’t use the deprecated ES 7.x connection format.
Explicit mappings are safer — define the type of every field explicitly rather than relying on dynamic mapping; avoid indexing unneeded fields to save storage.
text vs keyword — use text for full-text search (product names, descriptions); use keyword for exact match, sorting, and aggregations (category, status, tags).
The bool query is the foundation — combine must (affects score), filter (doesn’t affect score, faster), and should (optional, boosts score) for flexible queries.
filter is faster than must — conditions that don’t need to affect the score (category, status, price range) should go in filter, not must.
bulk() for mass indexing — far more efficient than a loop of individual index() calls; use a generator to save memory with very large data.
size=0 for pure aggregations — set size=0 when you only need aggregation results without documents; saves bandwidth and memory.
highlight — use it to show text snippets matching the query, great for search UX.
track_total_hits=True — enable it when you need the total count of matching documents for accurate pagination.
Elasticsearch as a search layer — use it alongside a primary database, not as a replacement; store data in PostgreSQL/MySQL and index it into Elasticsearch for search.

← Previous: MongoDB Next: Kafka →