Elasticsearch #

Elasticsearch is a distributed search engine built on Apache Lucene, optimized for full-text search, log analytics, and observability. Unlike ordinary databases that store data for retrieval, Elasticsearch indexes every word in a document for super-fast search even on datasets of billions of documents. In Rust, the elasticsearch crate from Elastic provides a complete async client. Since Elasticsearch communicates via REST API with JSON, all requests and responses use serde_json::Value — this makes the API very flexible, though slightly less type-safe than sqlx. This article covers everything from basic setup to advanced queries and sync patterns with the primary database.

Installation #

[dependencies]
elasticsearch = "8"
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
chrono = { version = "0.4", features = ["serde"] }

Elasticsearch Basic Concepts #

flowchart LR
    subgraph Elasticsearch
        IDX["Index\n(like a table)"]
        IDX --> D1["Document\n{_id, _source: {...}}"]
        IDX --> D2["Document"]
        IDX --> S["Shard 1..N\n(data partition)"]
    end

    subgraph Cluster
        N1["Node 1\n(master + data)"]
        N2["Node 2\n(data)"]
        N3["Node 3\n(data)"]
    end
RDBMS ConceptElasticsearch Concept
DatabaseCluster
TableIndex
RowDocument
ColumnField
SchemaMapping
SQL QueryQuery DSL (JSON)
Full-text searchInverted index (built-in)

Connections and Clients #

use elasticsearch::{
    auth::Credentials,
    cert::CertificateValidation,
    http::transport::{SingleNodeConnectionPool, TransportBuilder},
    Elasticsearch,
};
use url::Url;

fn buat_client(url: &str) -> Result<Elasticsearch, Box<dyn std::error::Error>> {
    let url = Url::parse(url)?;
    let pool = SingleNodeConnectionPool::new(url);
    let transport = TransportBuilder::new(pool)
        .disable_proxy()
        .build()?;
    Ok(Elasticsearch::new(transport))
}

// With authentication (Elastic Cloud or self-hosted with security)
fn buat_client_auth(
    url: &str,
    username: &str,
    password: &str,
) -> Result<Elasticsearch, Box<dyn std::error::Error>> {
    let url = Url::parse(url)?;
    let pool = SingleNodeConnectionPool::new(url);
    let creds = Credentials::Basic(username.to_string(), password.to_string());
    let transport = TransportBuilder::new(pool)
        .auth(creds)
        .cert_validation(CertificateValidation::None)  // for dev/testing
        .build()?;
    Ok(Elasticsearch::new(transport))
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = buat_client("http://localhost:9200")?;

    // Verify the connection
    let info = client.info().send().await?;
    let body: serde_json::Value = info.json().await?;
    println!("Elasticsearch: {}", body["version"]["number"].as_str().unwrap_or("?"));

    Ok(())
}

Mapping — Defining the Index Structure #

Mapping in Elasticsearch is similar to a schema in a database — it defines the type of every field:

use elasticsearch::{indices::IndicesCreateParts, Elasticsearch};
use serde_json::{json, Value};

async fn buat_index_artikel(
    client: &Elasticsearch,
) -> Result<(), Box<dyn std::error::Error>> {
    let mapping = json!({
        "settings": {
            "number_of_shards": 1,
            "number_of_replicas": 0,  // 0 for single-node dev
            "analysis": {
                "analyzer": {
                    "analyzer_indonesia": {
                        "type": "standard",
                        "stopwords": "_indonesian_"
                    }
                }
            }
        },
        "mappings": {
            "properties": {
                "judul": {
                    "type": "text",
                    "analyzer": "analyzer_indonesia",
                    "fields": {
                        "keyword": {  // sub-field for exact match and sorting
                            "type": "keyword"
                        }
                    }
                },
                "konten": {
                    "type": "text",
                    "analyzer": "analyzer_indonesia"
                },
                "tag": {
                    "type": "keyword"  // keyword: exact match, not analyzed
                },
                "pengguna_id": {
                    "type": "long"
                },
                "diterbitkan": {
                    "type": "boolean"
                },
                "views": {
                    "type": "long"
                },
                "dibuat_pada": {
                    "type": "date",
                    "format": "strict_date_optional_time||epoch_millis"
                }
            }
        }
    });

    // Create the index — use ignore_unavailable for idempotency
    let respons = client
        .indices()
        .create(IndicesCreateParts::Index("artikel"))
        .body(mapping)
        .send()
        .await?;

    if respons.status_code().is_success() {
        println!("Index 'artikel' created successfully");
    } else {
        let error: Value = respons.json().await?;
        println!("Index may already exist: {}", error["error"]["type"].as_str().unwrap_or("?"));
    }

    Ok(())
}

// Delete and recreate an index (for development)
async fn reset_index(
    client: &Elasticsearch,
    nama_index: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    use elasticsearch::indices::{IndicesDeleteParts, IndicesExistsParts};

    // Delete if it exists
    let ada = client
        .indices()
        .exists(IndicesExistsParts::Index(&[nama_index]))
        .send()
        .await?
        .status_code()
        .is_success();

    if ada {
        client
            .indices()
            .delete(IndicesDeleteParts::Index(&[nama_index]))
            .send()
            .await?;
        println!("Index '{}' deleted", nama_index);
    }

    Ok(())
}

Indexing Documents #

use elasticsearch::{IndexParts, BulkParts, Elasticsearch};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};

#[derive(Debug, Serialize, Deserialize, Clone)]
struct DokumenArtikel {
    pub id: i64,
    pub judul: String,
    pub konten: String,
    pub tag: Vec<String>,
    pub pengguna_id: i64,
    pub diterbitkan: bool,
    pub views: u64,
    pub dibuat_pada: String,  // ISO 8601 string
}

// Index a single document
async fn index_artikel(
    client: &Elasticsearch,
    artikel: &DokumenArtikel,
) -> Result<String, Box<dyn std::error::Error>> {
    let respons = client
        .index(IndexParts::IndexId("artikel", &artikel.id.to_string()))
        .body(artikel)
        .send()
        .await?;

    let body: Value = respons.json().await?;
    let result = body["result"].as_str().unwrap_or("unknown").to_string();
    println!("Document {}: {}", artikel.id, result); // "created" or "updated"
    Ok(result)
}

// Bulk indexing — far more efficient for many documents
async fn bulk_index_artikel(
    client: &Elasticsearch,
    artikel_list: &[DokumenArtikel],
) -> Result<u64, Box<dyn std::error::Error>> {
    let mut body: Vec<Value> = Vec::new();

    for artikel in artikel_list {
        // Each bulk operation consists of two lines: action + document
        body.push(json!({
            "index": {
                "_index": "artikel",
                "_id": artikel.id.to_string()
            }
        }));
        body.push(serde_json::to_value(artikel)?);
    }

    let respons = client
        .bulk(BulkParts::None)
        .body(body)
        .send()
        .await?;

    let result: Value = respons.json().await?;

    // Count the successful documents
    let berhasil = result["items"]
        .as_array()
        .map(|items| items.iter().filter(|item| item["index"]["error"].is_null()).count())
        .unwrap_or(0);

    println!("Bulk index: {}/{} succeeded", berhasil, artikel_list.len());
    Ok(berhasil as u64)
}

// Delete a document
async fn hapus_dokumen(
    client: &Elasticsearch,
    id: i64,
) -> Result<bool, Box<dyn std::error::Error>> {
    use elasticsearch::DeleteParts;

    let respons = client
        .delete(DeleteParts::IndexId("artikel", &id.to_string()))
        .send()
        .await?;

    Ok(respons.status_code().is_success())
}

Search — Searching Documents #

Basic Full-Text Search #

use elasticsearch::{SearchParts, Elasticsearch};
use serde_json::{json, Value};

async fn cari_artikel(
    client: &Elasticsearch,
    query: &str,
) -> Result<Vec<Value>, Box<dyn std::error::Error>> {
    let respons = client
        .search(SearchParts::Index(&["artikel"]))
        .body(json!({
            "query": {
                // multi_match: search several fields at once
                "multi_match": {
                    "query": query,
                    "fields": [
                        "judul^3",   // ^3 = boost: judul is 3x more important than konten
                        "konten",
                        "tag"
                    ],
                    "type": "best_fields",
                    "fuzziness": "AUTO"  // automatic typo tolerance
                }
            },
            "size": 20,
            "_source": ["id", "judul", "tag", "views", "dibuat_pada"]  // select fields
        }))
        .send()
        .await?;

    let body: Value = respons.json().await?;
    let hits = body["hits"]["hits"]
        .as_array()
        .cloned()
        .unwrap_or_default();

    println!("Total found: {}", body["hits"]["total"]["value"]);
    Ok(hits)
}

Bool Queries — Combining Conditions #

async fn cari_artikel_lanjutan(
    client: &Elasticsearch,
    query: &str,
    tag: Option<&str>,
    hanya_diterbitkan: bool,
    min_views: Option<u64>,
) -> Result<Vec<Value>, Box<dyn std::error::Error>> {
    let mut must: Vec<Value> = vec![];
    let mut filter: Vec<Value> = vec![];

    // must: conditions that affect the relevance score
    if !query.is_empty() {
        must.push(json!({
            "multi_match": {
                "query": query,
                "fields": ["judul^3", "konten"]
            }
        }));
    }

    // filter: conditions that don't affect the score (faster, cached)
    if hanya_diterbitkan {
        filter.push(json!({"term": {"diterbitkan": true}}));
    }
    if let Some(t) = tag {
        filter.push(json!({"term": {"tag": t}}));
    }
    if let Some(views) = min_views {
        filter.push(json!({"range": {"views": {"gte": views}}}));
    }

    // If there's no text query, show everything (match_all)
    if must.is_empty() {
        must.push(json!({"match_all": {}}));
    }

    let respons = client
        .search(SearchParts::Index(&["artikel"]))
        .body(json!({
            "query": {
                "bool": {
                    "must": must,
                    "filter": filter
                }
            },
            "sort": [
                {"_score": "desc"},
                {"dibuat_pada": "desc"}
            ],
            "size": 20
        }))
        .send()
        .await?;

    let body: Value = respons.json().await?;
    Ok(body["hits"]["hits"].as_array().cloned().unwrap_or_default())
}

Highlighting — Marking Matching Words #

async fn cari_dengan_highlight(
    client: &Elasticsearch,
    query: &str,
) -> Result<Vec<(String, String)>, Box<dyn std::error::Error>> {
    let respons = client
        .search(SearchParts::Index(&["artikel"]))
        .body(json!({
            "query": {
                "multi_match": {
                    "query": query,
                    "fields": ["judul", "konten"]
                }
            },
            "highlight": {
                "fields": {
                    "judul": {
                        "pre_tags": ["<strong>"],
                        "post_tags": ["</strong>"]
                    },
                    "konten": {
                        "pre_tags": ["<em>"],
                        "post_tags": ["</em>"],
                        "number_of_fragments": 3,
                        "fragment_size": 150
                    }
                }
            },
            "size": 10
        }))
        .send()
        .await?;

    let body: Value = respons.json().await?;
    let hasil: Vec<(String, String)> = body["hits"]["hits"]
        .as_array()
        .unwrap_or(&vec![])
        .iter()
        .map(|hit| {
            let judul = hit["highlight"]["judul"][0]
                .as_str()
                .or_else(|| hit["_source"]["judul"].as_str())
                .unwrap_or("")
                .to_string();
            let snippet = hit["highlight"]["konten"][0]
                .as_str()
                .unwrap_or("")
                .to_string();
            (judul, snippet)
        })
        .collect();

    Ok(hasil)
}

Aggregation — Analytics #

async fn statistik_artikel(
    client: &Elasticsearch,
) -> Result<Value, Box<dyn std::error::Error>> {
    let respons = client
        .search(SearchParts::Index(&["artikel"]))
        .body(json!({
            "size": 0,  // no individual documents needed — only aggregation
            "aggs": {
                "per_tag": {
                    "terms": {
                        "field": "tag",
                        "size": 10  // top 10 tags
                    },
                    "aggs": {
                        "total_views": {
                            "sum": {"field": "views"}
                        },
                        "rata_views": {
                            "avg": {"field": "views"}
                        }
                    }
                },
                "per_bulan": {
                    "date_histogram": {
                        "field": "dibuat_pada",
                        "calendar_interval": "month",
                        "format": "yyyy-MM"
                    },
                    "aggs": {
                        "jumlah_diterbitkan": {
                            "filter": {"term": {"diterbitkan": true}}
                        }
                    }
                },
                "views_stats": {
                    "stats": {"field": "views"}  // min, max, avg, sum, count
                }
            }
        }))
        .send()
        .await?;

    let body: Value = respons.json().await?;

    // Display the aggregations
    if let Some(per_tag) = body["aggregations"]["per_tag"]["buckets"].as_array() {
        println!("\n=== Top Tags ===");
        for bucket in per_tag {
            println!(
                "{}: {} articles, {} total views",
                bucket["key"].as_str().unwrap_or("?"),
                bucket["doc_count"],
                bucket["total_views"]["value"]
            );
        }
    }

    Ok(body["aggregations"].clone())
}

Pagination with search_after #

For large datasets, from/size becomes inefficient after page 100+. Use search_after for efficient deep pagination:

async fn cari_berhalaman(
    client: &Elasticsearch,
    query: &str,
    per_halaman: u64,
    search_after: Option<Vec<Value>>,  // cursor from the previous page
) -> Result<(Vec<Value>, Option<Vec<Value>>), Box<dyn std::error::Error>> {
    let mut body = json!({
        "query": {
            "multi_match": {
                "query": query,
                "fields": ["judul", "konten"]
            }
        },
        "sort": [
            {"_score": "desc"},
            {"_id": "asc"}  // tie-breaker required for search_after
        ],
        "size": per_halaman,
        // pit: point in time for page consistency
    });

    // Add search_after if present (next page)
    if let Some(cursor) = search_after {
        body["search_after"] = json!(cursor);
    }

    let respons = client
        .search(SearchParts::Index(&["artikel"]))
        .body(body)
        .send()
        .await?;

    let result: Value = respons.json().await?;
    let hits = result["hits"]["hits"]
        .as_array()
        .cloned()
        .unwrap_or_default();

    // Take the sort values from the last document as the next cursor
    let cursor_berikutnya = hits.last()
        .and_then(|hit| hit["sort"].as_array().cloned());

    Ok((hits, cursor_berikutnya))
}

Database → Elasticsearch Synchronization #

A common pattern in production architectures: the primary database (PostgreSQL/MySQL) as the source of truth, Elasticsearch for search:

sequenceDiagram
    participant API as API Server
    participant DB as PostgreSQL
    participant ES as Elasticsearch

    API->>DB: INSERT/UPDATE artikel
    DB->>API: OK + article data
    API->>ES: Index document (async)
    ES->>API: Acknowledged

    API->>ES: Search articles
    ES->>API: Search results (fast)

    API->>DB: Get article detail by ID
    DB->>API: Full data
use tokio::sync::mpsc;

#[derive(Debug, Clone)]
enum EventSinkronisasi {
    Index { id: i64, dokumen: DokumenArtikel },
    Hapus { id: i64 },
}

// Worker that listens for events and indexes into ES
async fn worker_sinkronisasi(
    client: elasticsearch::Elasticsearch,
    mut rx: mpsc::Receiver<EventSinkronisasi>,
) {
    while let Some(event) = rx.recv().await {
        match event {
            EventSinkronisasi::Index { id, dokumen } => {
                if let Err(e) = index_artikel(&client, &dokumen).await {
                    eprintln!("Failed to index article {}: {}", id, e);
                }
            }
            EventSinkronisasi::Hapus { id } => {
                if let Err(e) = hapus_dokumen(&client, id).await {
                    eprintln!("Failed to delete document {}: {}", id, e);
                }
            }
        }
    }
}

// In the API handler — send the event after the database operation
async fn handler_buat_artikel(
    tx: mpsc::Sender<EventSinkronisasi>,
    // ... article data from the request
) {
    // 1. Save to the database (source of truth)
    // let artikel = db.insert_artikel(...).await?;

    // 2. Send the event to be indexed into ES (fire and forget)
    let dokumen = DokumenArtikel {
        id: 1,
        judul: "Judul Artikel".to_string(),
        konten: "Full content...".to_string(),
        tag: vec!["rust".to_string()],
        pengguna_id: 1,
        diterbitkan: true,
        views: 0,
        dibuat_pada: chrono::Utc::now().to_rfc3339(),
    };

    let _ = tx.send(EventSinkronisasi::Index { id: 1, dokumen }).await;
}

Re-indexing — Updating All Documents #

async fn reindex_semua(
    client: &Elasticsearch,
    artikel_list: Vec<DokumenArtikel>,
) -> Result<(), Box<dyn std::error::Error>> {
    const BATCH_SIZE: usize = 500;

    // Delete the old index and recreate it
    reset_index(client, "artikel").await?;
    buat_index_artikel(client).await?;

    // Process in batches
    for (i, batch) in artikel_list.chunks(BATCH_SIZE).enumerate() {
        bulk_index_artikel(client, batch).await?;
        println!("Batch {}: {} documents indexed", i + 1, batch.len());
    }

    println!("Reindex complete: {} documents total", artikel_list.len());
    Ok(())
}

Summary #

  • Elasticsearch isn’t a database replacement — use it as a search layer on top of the primary database (PostgreSQL/MySQL). The database is the source of truth; Elasticsearch handles fast search queries.
  • Mapping matters for performance — define field types explicitly. text for full-text search, keyword for exact match and aggregation, date for time range queries.
  • Field boosting matters — "fields": ["judul^3", "konten"] makes title-field results 3x more relevant than content. Adjust weights based on the application domain.
  • filter vs must in bool queries — filter doesn’t affect the relevance score and is cached by Elasticsearch (faster). Use filter for binary conditions (active/inactive, date ranges), must for conditions that need scoring.
  • Bulk indexing for many documents — far more efficient than indexing one by one. Use a batch size of 500–1000 documents per request.
  • search_after for deep pagination — more efficient than from/size, which loads all documents from page 1. A unique sort field is required as a tie-breaker (e.g. _id).
  • Aggregation for analytics — terms, date_histogram, stats, range. Set size: 0 if you only need aggregations without documents.
  • Async synchronization with channels — send events to a worker that indexes into ES after successful database operations. Avoid blocking synchronization in the request hot path.
  • Re-indexing with zero downtime — create a new index, fill it from the database, then switch the alias. Don’t drop the active index before the new one is ready.

← Previous: MongoDB   Next: Kafka →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact