Elasticsearch #

Elasticsearch is a distributed search and analytics engine built on top of Apache Lucene. It excels at full-text search, log analytics (ELK Stack), and complex data searches — far surpassing what LIKE '%keyword%' can do in SQL. Go supports Elasticsearch through the official github.com/elastic/go-elasticsearch client.

Installation #

# The official Elasticsearch client — choose the version matching your ES server
go get github.com/elastic/go-elasticsearch/v8  # for ES 8.x
go get github.com/elastic/go-elasticsearch/v7  # for ES 7.x

Connecting to Elasticsearch #

import (
    "github.com/elastic/go-elasticsearch/v8"
    "github.com/elastic/go-elasticsearch/v8/esapi"
)

func connectES() (*elasticsearch.Client, error) {
    cfg := elasticsearch.Config{
        Addresses: []string{
            "http://localhost:9200",
            // add other nodes for a cluster
        },
        // For ES with authentication
        Username: "elastic",
        Password: "password",

        // Or with an API key
        // APIKey: "base6...ey",

        // Retry configuration
        RetryOnStatus: []int{502, 503, 504},
        MaxRetries:    3,
    }

    client, err := elasticsearch.NewClient(cfg)
    if err != nil {
        return nil, fmt.Errorf("create client: %w", err)
    }

    // Ping to verify
    res, err := client.Info()
    if err != nil {
        return nil, fmt.Errorf("ping: %w", err)
    }
    defer res.Body.Close()

    if res.IsError() {
        return nil, fmt.Errorf("info error: %s", res.Status())
    }

    fmt.Println("✓ Connected to Elasticsearch")
    return client, nil
}

Main Field Types in Elasticsearch #

Before defining an index mapping, understand the main Lucene field types below:

Field TypeMain CharacteristicsSupported OperationsExample Uses
textAnalyzed (tokenization, stemming, stopwords)Full-text searchProduct descriptions, article content
keywordStored exactly as-is (exact value)Exact filters, sorting, aggregationOrder status, tags, category codes
integer / longOptimized numeric data typesRange queries (price, stock)Inventory counts, item prices
dateSupports various date formatsTime ordering, date range filterscreated_at, updated_at
geo_pointLatitude and longitude coordinates (lat/lon)Geographic distance measurementsBranch locations, driver coordinates

The Data Write Replication Flow (Indexing Write Path) #

When a Go program performs an index operation (writing a new document), the coordination between shards in the Elasticsearch cluster runs as follows:

flowchart TD
    Client["Go Application"] -->|"1. Send Index Request"| Coord["Coordinating Node"]
    Coord -->|"2. Determine Shard via Hash (ID)"| Primary["Primary Shard (Node A)"]
    
    subgraph replication["Parallel Replication"]
        Primary -->|"3. Write Local Data"| Storage["Shard Storage"]
        Primary -->|"4. Send Replication"| Replica1["Replica Shard 1 (Node B)"]
        Primary -->|"4. Send Replication"| Replica2["Replica Shard 2 (Node C)"]
    end
    
    Replica1 -->|"5. Confirm Success"| Primary
    Replica2 -->|"5. Confirm Success"| Primary
    Primary -->|"6. Confirm Completion"| Coord
    Coord -->|"7. Return HTTP 201 Created"| Client

Index Mapping #

Mapping defines the field types in an index — important for search and aggregation:

const productMapping = `{
    "settings": {
        "number_of_shards": 1,
        "number_of_replicas": 0,
        "analysis": {
            "analyzer": {
                "indonesian_analyzer": {
                    "type": "standard",
                    "stopwords": "_indonesian_"
                }
            }
        }
    },
    "mappings": {
        "properties": {
            "name": {
                "type": "text",
                "analyzer": "indonesian_analyzer",
                "fields": {
                    "keyword": { "type": "keyword" }
                }
            },
            "description": {
                "type": "text",
                "analyzer": "indonesian_analyzer"
            },
            "category": { "type": "keyword" },
            "brand":    { "type": "keyword" },
            "tags":     { "type": "keyword" },
            "price":    { "type": "double" },
            "stock":    { "type": "integer" },
            "rating":   { "type": "float" },
            "is_active": { "type": "boolean" },
            "created_at": { "type": "date" },
            "specs": { "type": "object" }
        }
    }
}`

func createIndex(es *elasticsearch.Client, indexName string) error {
    res, err := es.Indices.Create(
        indexName,
        es.Indices.Create.WithBody(strings.NewReader(productMapping)),
    )
    if err != nil {
        return err
    }
    defer res.Body.Close()

    if res.IsError() {
        var e map[string]interface{}
        json.NewDecoder(res.Body).Decode(&e)
        // The index already exists — no problem
        if e["error"].(map[string]interface{})["type"] == "resource_already_exists_exception" {
            return nil
        }
        return fmt.Errorf("create index: %s", res.Status())
    }
    return nil
}

Indexing — Storing Documents #

type Product struct {
    ID          string            `json:"id"`
    Name        string            `json:"name"`
    Description string            `json:"description"`
    Category    string            `json:"category"`
    Brand       string            `json:"brand"`
    Tags        []string          `json:"tags"`
    Price       float64           `json:"price"`
    Stock       int               `json:"stock"`
    Rating      float64           `json:"rating"`
    IsActive    bool              `json:"is_active"`
    Specs       map[string]string `json:"specs,omitempty"`
    CreatedAt   time.Time         `json:"created_at"`
}

func indexProduct(es *elasticsearch.Client, indexName string, p Product) error {
    data, err := json.Marshal(p)
    if err != nil {
        return err
    }

    res, err := es.Index(
        indexName,
        bytes.NewReader(data),
        es.Index.WithDocumentID(p.ID),
        es.Index.WithRefresh("true"), // immediately visible — dev/test only
    )
    if err != nil {
        return fmt.Errorf("index: %w", err)
    }
    defer res.Body.Close()

    if res.IsError() {
        return fmt.Errorf("index error: %s", res.Status())
    }
    return nil
}

Bulk Indexing — Efficiently Inserting Many Documents #

For indexing large amounts of data, use the Bulk API:

func bulkIndex(es *elasticsearch.Client, indexName string, products []Product) error {
    var buf bytes.Buffer

    for _, p := range products {
        // Each document needs two lines: action and data
        meta := fmt.Sprintf(`{"index":{"_index":%q,"_id":%q}}%s`,
            indexName, p.ID, "\n")
        buf.WriteString(meta)

        data, _ := json.Marshal(p)
        buf.Write(data)
        buf.WriteByte('\n')
    }

    res, err := es.Bulk(bytes.NewReader(buf.Bytes()),
        es.Bulk.WithIndex(indexName),
        es.Bulk.WithRefresh("true"),
    )
    if err != nil {
        return fmt.Errorf("bulk: %w", err)
    }
    defer res.Body.Close()

    if res.IsError() {
        return fmt.Errorf("bulk error: %s", res.Status())
    }

    var result map[string]interface{}
    json.NewDecoder(res.Body).Decode(&result)

    if result["errors"].(bool) {
        return fmt.Errorf("there were errors in the bulk indexing")
    }

    items := result["items"].([]interface{})
    fmt.Printf("Bulk index: %d documents processed\n", len(items))
    return nil
}

Search — Finding Documents #

Basic Full-Text Search #

func search(es *elasticsearch.Client, indexName, keyword string) ([]Product, error) {
    query := map[string]interface{}{
        "query": map[string]interface{}{
            "multi_match": map[string]interface{}{
                "query":  keyword,
                "fields": []string{"name^3", "description", "tags^2"},
                // ^3 = boosts the name 3x more
            },
        },
        "highlight": map[string]interface{}{
            "fields": map[string]interface{}{
                "name":        map[string]interface{}{},
                "description": map[string]interface{}{},
            },
        },
    }

    return executeSearch(es, indexName, query)
}

Bool Queries — Complex Filters #

func advancedSearch(es *elasticsearch.Client, indexName string, params SearchParams) (SearchResult, error) {
    // Bool query: must (AND), should (OR), must_not (NOT), filter (AND, no scoring)
    boolQuery := map[string]interface{}{
        "must": []interface{}{},
        "filter": []interface{}{
            map[string]interface{}{"term": map[string]interface{}{"is_active": true}},
        },
        "must_not": []interface{}{},
        "should":   []interface{}{},
    }

    // Full-text search on name and description
    if params.Keyword != "" {
        boolQuery["must"] = append(boolQuery["must"].([]interface{}),
            map[string]interface{}{
                "multi_match": map[string]interface{}{
                    "query":     params.Keyword,
                    "fields":    []string{"name^3", "description", "tags^2"},
                    "type":      "best_fields",
                    "fuzziness": "AUTO", // typo tolerance
                },
            },
        )
    }

    // Category filter (exact match)
    if params.Category != "" {
        boolQuery["filter"] = append(boolQuery["filter"].([]interface{}),
            map[string]interface{}{"term": map[string]interface{}{"category": params.Category}},
        )
    }

    // Price range filter
    if params.MinPrice > 0 || params.MaxPrice > 0 {
        priceRange := map[string]interface{}{}
        if params.MinPrice > 0 {
            priceRange["gte"] = params.MinPrice
        }
        if params.MaxPrice > 0 {
            priceRange["lte"] = params.MaxPrice
        }
        boolQuery["filter"] = append(boolQuery["filter"].([]interface{}),
            map[string]interface{}{"range": map[string]interface{}{"price": priceRange}},
        )
    }

    // In-stock filter
    if params.InStockOnly {
        boolQuery["filter"] = append(boolQuery["filter"].([]interface{}),
            map[string]interface{}{"range": map[string]interface{}{"stock": map[string]interface{}{"gt": 0}}},
        )
    }

    // Sort
    sortField := "created_at"
    sortOrder := "desc"
    if params.SortBy == "price_asc" {
        sortField, sortOrder = "price", "asc"
    } else if params.SortBy == "price_desc" {
        sortField, sortOrder = "price", "desc"
    } else if params.SortBy == "rating" {
        sortField, sortOrder = "rating", "desc"
    }

    query := map[string]interface{}{
        "query": map[string]interface{}{"bool": boolQuery},
        "sort":  []interface{}{map[string]interface{}{sortField: sortOrder}},
        "from":  (params.Page - 1) * params.PerPage,
        "size":  params.PerPage,
        "highlight": map[string]interface{}{
            "fields": map[string]interface{}{
                "name":        map[string]interface{}{},
                "description": map[string]interface{}{"fragment_size": 150},
            },
            "pre_tags":  []string{"<mark>"},
            "post_tags": []string{"</mark>"},
        },
        "aggs": map[string]interface{}{
            "by_category": map[string]interface{}{
                "terms": map[string]interface{}{"field": "category", "size": 20},
            },
            "price_range": map[string]interface{}{
                "range": map[string]interface{}{
                    "field": "price",
                    "ranges": []interface{}{
                        map[string]interface{}{"key": "< 500k", "to": 500_000},
                        map[string]interface{}{"key": "500k - 2m", "from": 500_000, "to": 2_000_000},
                        map[string]interface{}{"key": "2m - 10m", "from": 2_000_000, "to": 10_000_000},
                        map[string]interface{}{"key": "> 10m", "from": 10_000_000},
                    },
                },
            },
            "avg_price": map[string]interface{}{
                "avg": map[string]interface{}{"field": "price"},
            },
        },
    }

    return executeAdvancedSearch(es, indexName, query)
}

Aggregation — Analytics #

func productAnalytics(es *elasticsearch.Client, indexName string) error {
    query := map[string]interface{}{
        "size": 0, // no documents needed, only aggregations
        "aggs": map[string]interface{}{
            "categories": map[string]interface{}{
                "terms": map[string]interface{}{
                    "field": "category",
                    "size":  10,
                },
                "aggs": map[string]interface{}{
                    "avg_price": map[string]interface{}{
                        "avg": map[string]interface{}{"field": "price"},
                    },
                    "total_stock": map[string]interface{}{
                        "sum": map[string]interface{}{"field": "stock"},
                    },
                    "avg_rating": map[string]interface{}{
                        "avg": map[string]interface{}{"field": "rating"},
                    },
                },
            },
            "price_stats": map[string]interface{}{
                "extended_stats": map[string]interface{}{"field": "price"},
            },
            "products_per_day": map[string]interface{}{
                "date_histogram": map[string]interface{}{
                    "field":             "created_at",
                    "calendar_interval": "day",
                    "format":            "yyyy-MM-dd",
                },
            },
        },
    }

    data, _ := json.Marshal(query)
    res, err := es.Search(
        es.Search.WithIndex(indexName),
        es.Search.WithBody(bytes.NewReader(data)),
    )
    if err != nil {
        return err
    }
    defer res.Body.Close()

    var result map[string]interface{}
    json.NewDecoder(res.Body).Decode(&result)

    aggs := result["aggregations"].(map[string]interface{})
    categories := aggs["categories"].(map[string]interface{})["buckets"].([]interface{})

    fmt.Println("=== Analytics per Category ===")
    for _, bucket := range categories {
        b := bucket.(map[string]interface{})
        fmt.Printf("  %-15s: %v products, avg Rp%.0f, stock %v, rating %.1f\n",
            b["key"],
            b["doc_count"],
            b["avg_price"].(map[string]interface{})["value"],
            b["total_stock"].(map[string]interface{})["value"],
            b["avg_rating"].(map[string]interface{})["value"],
        )
    }
    return nil
}

Helper Functions #

type SearchParams struct {
    Keyword     string
    Category    string
    MinPrice    float64
    MaxPrice    float64
    InStockOnly bool
    SortBy      string
    Page        int
    PerPage     int
}

type SearchResult struct {
    Total    int64
    Products []ProductHit
    Aggs     map[string]interface{}
}

type ProductHit struct {
    Product    Product
    Score      float64
    Highlights map[string][]string
}

func executeSearch(es *elasticsearch.Client, indexName string, query map[string]interface{}) ([]Product, error) {
    data, _ := json.Marshal(query)
    res, err := es.Search(
        es.Search.WithIndex(indexName),
        es.Search.WithBody(bytes.NewReader(data)),
    )
    if err != nil {
        return nil, err
    }
    defer res.Body.Close()

    if res.IsError() {
        return nil, fmt.Errorf("search error: %s", res.Status())
    }

    var result struct {
        Hits struct {
            Total struct{ Value int64 }
            Hits  []struct {
                Source Product `json:"_source"`
                Score  float64 `json:"_score"`
            }
        }
    }
    json.NewDecoder(res.Body).Decode(&result)

    products := make([]Product, len(result.Hits.Hits))
    for i, hit := range result.Hits.Hits {
        products[i] = hit.Source
    }
    return products, nil
}

func executeAdvancedSearch(es *elasticsearch.Client, indexName string, query map[string]interface{}) (SearchResult, error) {
    data, _ := json.Marshal(query)
    res, err := es.Search(
        es.Search.WithIndex(indexName),
        es.Search.WithBody(bytes.NewReader(data)),
    )
    if err != nil {
        return SearchResult{}, err
    }
    defer res.Body.Close()

    var raw map[string]interface{}
    json.NewDecoder(res.Body).Decode(&raw)

    hits := raw["hits"].(map[string]interface{})
    total := int64(hits["total"].(map[string]interface{})["value"].(float64))

    var products []ProductHit
    for _, hit := range hits["hits"].([]interface{}) {
        h := hit.(map[string]interface{})
        sourceBytes, _ := json.Marshal(h["_source"])
        var p Product
        json.Unmarshal(sourceBytes, &p)

        ph := ProductHit{
            Product: p,
            Score:   h["_score"].(float64),
        }

        // Extract highlights
        if hl, ok := h["highlight"].(map[string]interface{}); ok {
            ph.Highlights = make(map[string][]string)
            for field, frags := range hl {
                for _, frag := range frags.([]interface{}) {
                    ph.Highlights[field] = append(ph.Highlights[field], frag.(string))
                }
            }
        }
        products = append(products, ph)
    }

    return SearchResult{
        Total:    total,
        Products: products,
        Aggs:     raw["aggregations"].(map[string]interface{}),
    }, nil
}

Complete Example Program #

package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "log"
    "strings"
    "time"

    "github.com/elastic/go-elasticsearch/v8"
)

const indexName = "products_demo"

func main() {
    es, err := elasticsearch.NewDefaultClient()
    if err != nil {
        log.Fatal(err)
    }

    // Check the connection
    res, err := es.Info()
    if err != nil {
        log.Fatal("Cannot connect to Elasticsearch:", err)
    }
    defer res.Body.Close()
    fmt.Println("✓ Connected to Elasticsearch")

    // Delete and recreate the index for the demo
    es.Indices.Delete([]string{indexName})
    if err := createIndex(es, indexName); err != nil {
        log.Fatal("Create index:", err)
    }
    fmt.Println("✓ Index created:", indexName)

    // Bulk index products
    products := []Product{
        {
            ID: "1", Name: "Pro Gaming Laptop",
            Description: "High-end gaming laptop with an RTX 4090 GPU",
            Category: "electronics", Brand: "ASUS",
            Tags: []string{"laptop", "gaming", "rtx"}, Price: 35_000_000, Stock: 5, Rating: 4.8,
            IsActive: true, CreatedAt: time.Now(),
        },
        {
            ID: "2", Name: "Ultrabook Laptop",
            Description: "Thin and light laptop for professionals",
            Category: "electronics", Brand: "Dell",
            Tags: []string{"laptop", "ultrabook", "thin"}, Price: 18_000_000, Stock: 12, Rating: 4.5,
            IsActive: true, CreatedAt: time.Now(),
        },
        {
            ID: "3", Name: "RGB Gaming Mouse",
            Description: "Gaming mouse with a high-precision sensor and RGB lighting",
            Category: "electronics", Brand: "Logitech",
            Tags: []string{"mouse", "gaming", "rgb"}, Price: 750_000, Stock: 45, Rating: 4.6,
            IsActive: true, CreatedAt: time.Now(),
        },
        {
            ID: "4", Name: "TKL Mechanical Keyboard",
            Description: "Tenkeyless mechanical keyboard with linear red switches",
            Category: "electronics", Brand: "Keychron",
            Tags: []string{"keyboard", "mechanical"}, Price: 1_200_000, Stock: 30, Rating: 4.7,
            IsActive: true, CreatedAt: time.Now(),
        },
        {
            ID: "5", Name: "Oversize Gaming Tee",
            Description: "Oversize gaming t-shirt made of premium 100% cotton",
            Category: "fashion", Brand: "GameWear",
            Tags: []string{"shirt", "gaming", "oversize"}, Price: 150_000, Stock: 200, Rating: 4.3,
            IsActive: true, CreatedAt: time.Now(),
        },
        {
            ID: "6", Name: "4K 144Hz Monitor",
            Description: "4K gaming monitor with a 144Hz refresh rate and HDR",
            Category: "electronics", Brand: "LG",
            Tags: []string{"monitor", "4k", "gaming"}, Price: 12_000_000, Stock: 8, Rating: 4.9,
            IsActive: true, CreatedAt: time.Now(),
        },
    }

    if err := bulkIndex(es, indexName, products); err != nil {
        log.Fatal("Bulk index:", err)
    }

    // Wait a moment so the data gets indexed
    time.Sleep(1 * time.Second)

    // Search: full-text
    fmt.Println("\n=== Full-Text Search: 'laptop gaming' ===")
    results, err := search(es, indexName, "laptop gaming")
    if err != nil {
        log.Println(err)
    } else {
        for _, p := range results {
            fmt.Printf("  %-25s Rp%10.0f ★%.1f\n", p.Name, p.Price, p.Rating)
        }
    }

    // Advanced search: filter + sort
    fmt.Println("\n=== Advanced Search: electronics, Rp500k-Rp15m, sort by price ===")
    adv, err := advancedSearch(es, indexName, SearchParams{
        Category: "electronics",
        MinPrice: 500_000,
        MaxPrice: 15_000_000,
        SortBy:   "price_asc",
        Page:     1,
        PerPage:  10,
    })
    if err != nil {
        log.Println(err)
    } else {
        fmt.Printf("  Total: %d results\n", adv.Total)
        for _, hit := range adv.Products {
            fmt.Printf("  %-25s Rp%10.0f (score: %.2f)\n",
                hit.Product.Name, hit.Product.Price, hit.Score)
            for field, frags := range hit.Highlights {
                fmt.Printf("    [%s] %s\n", field, strings.Join(frags, " | "))
            }
        }
    }

    // Aggregation analytics
    fmt.Println()
    if err := productAnalytics(es, indexName); err != nil {
        log.Println(err)
    }

    // Delete the index after the demo
    es.Indices.Delete([]string{indexName})
}

Summary #

  • elasticsearch.NewClient(cfg) for the connection; always Ping/check Info() to verify an active connection.
  • Mapping defines field types before indexing — text for full-text search, keyword for exact-match filters/aggregation.
  • The Bulk API for large-scale indexing — far more efficient than indexing one by one.
  • multi_match to search across many fields at once; use ^ to boost (e.g. "name^3").
  • Bool queries with must, should, filter, must_not — filter doesn’t affect the relevance score.
  • fuzziness: "AUTO" for typo tolerance — very useful for product search boxes.
  • Highlighting to show users the matching context — display the matching text with markup.
  • Aggregation for faceted search (category/price-range filters) and analytics without separate queries.
  • "size": 0 for analytics-only queries — no documents returned, only aggregations.
  • Index aliases for zero-downtime reindexing — add an alias to the new index, then switch the alias from the old to the new one.

← Previous: MongoDB   Next: Kafka →

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