Elasticsearch #

Elasticsearch is a distributed search and analytics engine built on Apache Lucene — designed for fast full-text search, real-time log analytics, and large-scale semi-structured data storage. Unlike other databases that have dedicated Dart drivers, Elasticsearch is accessed through a standard REST API over HTTP — which means the Dart http package is enough, with no additional driver dependencies. This article covers interacting with Elasticsearch from Dart effectively, from index management to complex query DSL.

Elasticsearch Basic Concepts #

Concept	Elasticsearch	Equivalent elsewhere
Index	A collection of documents with the same mapping	Table (SQL) / Collection (MongoDB)
Document	One JSON record in an index	Row (SQL) / Document (MongoDB)
Field	An attribute within a document	Column (SQL) / Field (MongoDB)
Shard	A horizontal slice of an index	Partition
Replica	A copy of a shard for failover	Replica
Mapping	The definition of each field’s type	Schema (SQL)
Query DSL	JSON for defining searches	SQL WHERE clause
Aggregation	Statistical calculations	GROUP BY + SQL aggregate functions

Client Setup #

Since Elasticsearch uses a REST API, no dedicated driver package is required. Create a reusable HTTP client wrapper:

dart pub add http

import 'dart:convert';
import 'package:http/http.dart' as http;

class ElasticsearchClient {
  final String baseUrl;
  final Map<String, String> _headers;

  ElasticsearchClient({
    required this.baseUrl,  // example: 'http://localhost:9200'
    String? username,
    String? password,
    String? apiKey,
  }) : _headers = {
    'Content-Type': 'application/json',
    'Accept': 'application/json',
    if (username != null && password != null)
      'Authorization': 'Basic ${base64Encode(utf8.encode('$username:$password'))}',
    if (apiKey != null)
      'Authorization': 'ApiKey $apiKey',
  };

  Uri _uri(String path, [Map<String, String>? params]) {
    final uri = Uri.parse('$baseUrl$path');
    return params != null ? uri.replace(queryParameters: params) : uri;
  }

  Future<Map<String, dynamic>> get(String path, [Map<String, String>? params]) async {
    final response = await http.get(_uri(path, params), headers: _headers);
    return _tanganiResponse(response);
  }

  Future<Map<String, dynamic>> post(String path, Map<String, dynamic> body) async {
    final response = await http.post(
      _uri(path),
      headers: _headers,
      body: jsonEncode(body),
    );
    return _tanganiResponse(response);
  }

  Future<Map<String, dynamic>> put(String path, Map<String, dynamic> body) async {
    final response = await http.put(
      _uri(path),
      headers: _headers,
      body: jsonEncode(body),
    );
    return _tanganiResponse(response);
  }

  Future<Map<String, dynamic>> delete(String path) async {
    final response = await http.delete(_uri(path), headers: _headers);
    return _tanganiResponse(response);
  }

  Map<String, dynamic> _tanganiResponse(http.Response response) {
    final body = jsonDecode(response.body) as Map<String, dynamic>;
    if (response.statusCode >= 400) {
      final error = body['error'] as Map<String, dynamic>?;
      throw ElasticsearchException(
        statusCode: response.statusCode,
        tipe: error?['type'] as String? ?? 'unknown_error',
        alasan: error?['reason'] as String? ?? response.body,
      );
    }
    return body;
  }
}

class ElasticsearchException implements Exception {
  final int statusCode;
  final String tipe;
  final String alasan;

  const ElasticsearchException({
    required this.statusCode,
    required this.tipe,
    required this.alasan,
  });

  @override
  String toString() => 'ElasticsearchException[$statusCode] $tipe: $alasan';
}

Index Management #

Creating an Index with a Mapping #

Mappings define the type of every field — Elasticsearch can auto-detect types, but explicit is better for full control:

Future<void> buatIndexProduk(ElasticsearchClient es) async {
  final mapping = {
    'settings': {
      'number_of_shards': 1,      // shards for small-medium production
      'number_of_replicas': 1,    // one replica for HA
      'analysis': {
        'analyzer': {
          'bahasa_indonesia': {
            'type': 'custom',
            'tokenizer': 'standard',
            'filter': ['lowercase', 'asciifolding'],
          }
        }
      }
    },
    'mappings': {
      'properties': {
        'nama': {
          'type': 'text',
          'analyzer': 'bahasa_indonesia',
          'fields': {
            'keyword': {'type': 'keyword'},  // for sorting and exact matches
          }
        },
        'deskripsi': {'type': 'text', 'analyzer': 'bahasa_indonesia'},
        'harga': {'type': 'double'},
        'stok': {'type': 'integer'},
        'kategori': {'type': 'keyword'},    // keyword = exact match, not tokenized
        'tag': {'type': 'keyword'},
        'aktif': {'type': 'boolean'},
        'dibuatPada': {'type': 'date'},
        'lokasi': {'type': 'geo_point'},    // for location-based search
        'spesifikasi': {'type': 'object'},  // nested JSON object
      }
    }
  };

  try {
    await es.put('/produk', mapping);
    print('Product index created successfully');
  } on ElasticsearchException catch (e) {
    if (e.statusCode == 400 && e.tipe == 'resource_already_exists_exception') {
      print('Index already exists, skipping creation');
    } else {
      rethrow;
    }
  }
}

// Check whether an index exists
Future<bool> indexAda(ElasticsearchClient es, String namaIndex) async {
  try {
    await es.get('/$namaIndex');
    return true;
  } on ElasticsearchException catch (e) {
    if (e.statusCode == 404) return false;
    rethrow;
  }
}

// Delete an index
Future<void> hapusIndex(ElasticsearchClient es, String namaIndex) async {
  await es.delete('/$namaIndex');
  print('Index $namaIndex deleted');
}

Document CRUD #

Index (Insert/Upsert) #

Future<void> contohIndex(ElasticsearchClient es) async {
  // Index with a specified ID — upsert (create or replace)
  final hasilDenganId = await es.put('/produk/_doc/P001', {
    'nama': 'Laptop Gaming ASUS ROG',
    'deskripsi': 'High-performance gaming laptop with an RTX 4070',
    'harga': 18_500_000.0,
    'stok': 5,
    'kategori': 'laptop',
    'tag': ['gaming', 'laptop', 'asus', 'rog'],
    'aktif': true,
    'dibuatPada': DateTime.now().toUtc().toIso8601String(),
    'spesifikasi': {
      'ram': '16GB DDR5',
      'storage': '1TB NVMe',
      'gpu': 'RTX 4070 8GB',
    },
  });
  print('Index result: ${hasilDenganId['result']}'); // created or updated

  // Index without an ID — Elasticsearch generates one automatically
  final hasilTanpaId = await es.post('/produk/_doc', {
    'nama': 'Mouse Gaming Logitech G502',
    'harga': 850_000.0,
    'stok': 20,
    'kategori': 'mouse',
    'aktif': true,
    'dibuatPada': DateTime.now().toUtc().toIso8601String(),
  });
  print('Created ID: ${hasilTanpaId['_id']}');

  // _create — only create if it doesn't exist (fails if the ID already exists)
  await es.put('/produk/_create/P999', {
    'nama': 'Produk Baru',
    'harga': 100_000.0,
    'aktif': true,
  });
}

Getting Documents #

Future<Map<String, dynamic>?> ambilProduk(
    ElasticsearchClient es, String id) async {
  try {
    final hasil = await es.get('/produk/_doc/$id');
    if (hasil['found'] as bool) {
      return hasil['_source'] as Map<String, dynamic>;
    }
    return null;
  } on ElasticsearchException catch (e) {
    if (e.statusCode == 404) return null;
    rethrow;
  }
}

// Multi-get — fetch several documents at once
Future<List<Map<String, dynamic>>> ambilBanyak(
    ElasticsearchClient es, List<String> ids) async {
  final hasil = await es.post('/produk/_mget', {
    'ids': ids,
  });

  return (hasil['docs'] as List)
      .where((doc) => doc['found'] as bool)
      .map((doc) => doc['_source'] as Map<String, dynamic>)
      .toList();
}

Updating Documents #

Future<void> contohUpdate(ElasticsearchClient es) async {
  // Partial update with the _update endpoint
  await es.post('/produk/_update/P001', {
    'doc': {
      'harga': 17_500_000.0,
      'stok': 3,
      'diubahPada': DateTime.now().toUtc().toIso8601String(),
    }
  });

  // Update with a script — for increment operations
  await es.post('/produk/_update/P001', {
    'script': {
      'source': 'ctx._source.stok -= params.qty',
      'lang': 'painless',
      'params': {'qty': 1},
    }
  });

  // Upsert — update if it exists, insert if it doesn't
  await es.post('/produk/_update/P002', {
    'doc': {'stok': 0, 'aktif': false},
    'upsert': {  // used when the document doesn't exist yet
      'nama': 'Produk Default',
      'stok': 0,
      'aktif': false,
    }
  });
}

Deleting Documents #

Future<void> hapusProduk(ElasticsearchClient es, String id) async {
  await es.delete('/produk/_doc/$id');
  print('Product $id deleted');
}

// Delete by query
Future<void> hapusByQuery(ElasticsearchClient es) async {
  final hasil = await es.post('/produk/_delete_by_query', {
    'query': {
      'term': {'aktif': false}
    }
  });
  print('Deleted: ${hasil['deleted']} documents');
}

Query DSL — Powerful Search #

Basic Queries #

Future<List<Map<String, dynamic>>> cariProduk(
    ElasticsearchClient es, String keyword) async {
  final body = {
    'query': {
      'multi_match': {
        'query': keyword,
        'fields': ['nama^2', 'deskripsi', 'tag'],  // ^2 = boost nama 2x
        'type': 'best_fields',
        'fuzziness': 'AUTO',  // typo tolerance: 'lapop' → 'laptop'
      }
    },
    'size': 20,
    'from': 0,
    'highlight': {  // show the matching parts
      'fields': {
        'nama': {},
        'deskripsi': {'fragment_size': 150},
      }
    },
    '_source': ['nama', 'harga', 'kategori', 'stok'],  // choose which fields to return
  };

  final hasil = await es.post('/produk/_search', body);
  final hits = hasil['hits']['hits'] as List;

  return hits.map((hit) => {
    ...hit['_source'] as Map<String, dynamic>,
    '_id': hit['_id'],
    '_score': hit['_score'],
    'highlight': hit['highlight'] ?? {},
  }).toList();
}

Bool Queries — Combining Conditions #

Future<List<Map<String, dynamic>>> cariDenganFilter(
  ElasticsearchClient es, {
  String? keyword,
  String? kategori,
  double? hargaMin,
  double? hargaMaks,
  List<String>? tags,
  int halaman = 1,
  int perHalaman = 20,
}) async {
  final mustClauses = <Map<String, dynamic>>[];
  final filterClauses = <Map<String, dynamic>>[];

  // Full-text search — affects the relevance score
  if (keyword != null && keyword.isNotEmpty) {
    mustClauses.add({
      'multi_match': {
        'query': keyword,
        'fields': ['nama^3', 'deskripsi', 'tag'],
        'fuzziness': 'AUTO',
      }
    });
  }

  // Filters — don't affect the score, faster because they're cached
  filterClauses.add({'term': {'aktif': true}});

  if (kategori != null) {
    filterClauses.add({'term': {'kategori': kategori}});
  }

  if (hargaMin != null || hargaMaks != null) {
    final range = <String, dynamic>{};
    if (hargaMin != null) range['gte'] = hargaMin;
    if (hargaMaks != null) range['lte'] = hargaMaks;
    filterClauses.add({'range': {'harga': range}});
  }

  if (tags != null && tags.isNotEmpty) {
    filterClauses.add({'terms': {'tag': tags}});
  }

  final query = mustClauses.isEmpty && filterClauses.isEmpty
      ? {'match_all': {}}
      : {
          'bool': {
            if (mustClauses.isNotEmpty) 'must': mustClauses,
            if (filterClauses.isNotEmpty) 'filter': filterClauses,
          }
        };

  final hasil = await es.post('/produk/_search', {
    'query': query,
    'sort': keyword != null
        ? ['_score', {'harga': 'asc'}]  // relevance first, then cheaper prices
        : [{'dibuatPada': 'desc'}],      // without a keyword: newest first
    'size': perHalaman,
    'from': (halaman - 1) * perHalaman,
    'track_total_hits': true,
  });

  final total = hasil['hits']['total']['value'] as int;
  print('Total: $total results');

  return (hasil['hits']['hits'] as List)
      .map((h) => {...h['_source'] as Map<String, dynamic>, '_id': h['_id']})
      .toList();
}

Future<Map<String, dynamic>> statistikProduk(ElasticsearchClient es) async {
  final hasil = await es.post('/produk/_search', {
    'size': 0,  // no documents needed, only aggregations
    'query': {'term': {'aktif': true}},
    'aggs': {
      // Count per category (facet)
      'perKategori': {
        'terms': {
          'field': 'kategori',
          'size': 20,
          'order': {'_count': 'desc'},
        }
      },
      // Price statistics
      'statsHarga': {
        'stats': {'field': 'harga'}
      },
      // Average price per category
      'hargaRataPerKategori': {
        'terms': {'field': 'kategori'},
        'aggs': {
          'rataHarga': {'avg': {'field': 'harga'}},
          'totalStok': {'sum': {'field': 'stok'}},
        }
      },
      // Price histogram — product distribution per price range
      'distribusiHarga': {
        'range': {
          'field': 'harga',
          'ranges': [
            {'to': 500_000},
            {'from': 500_000, 'to': 2_000_000},
            {'from': 2_000_000, 'to': 10_000_000},
            {'from': 10_000_000},
          ]
        }
      },
    }
  });

  final aggs = hasil['aggregations'] as Map<String, dynamic>;

  // Process the aggregation results
  final kategoriBuckets = (aggs['perKategori']['buckets'] as List)
      .map((b) => {'kategori': b['key'], 'jumlah': b['doc_count']})
      .toList();

  final statsHarga = aggs['statsHarga'] as Map<String, dynamic>;

  return {
    'perKategori': kategoriBuckets,
    'hargaMin': statsHarga['min'],
    'hargaMaks': statsHarga['max'],
    'hargaRata': statsHarga['avg'],
    'totalProduk': statsHarga['count'],
  };
}

Bulk Operations #

To index many documents at once, use the Bulk API — far more efficient than individual requests:

Future<void> bulkIndex(
    ElasticsearchClient es, List<Map<String, dynamic>> produkList) async {
  // The Bulk API uses NDJSON format (newline-delimited JSON)
  // Each operation: one action line + one document line
  final buffer = StringBuffer();

  for (final produk in produkList) {
    final id = produk['id'] as String?;

    // Action line
    if (id != null) {
      buffer.writeln(jsonEncode({'index': {'_index': 'produk', '_id': id}}));
    } else {
      buffer.writeln(jsonEncode({'index': {'_index': 'produk'}}));
    }

    // Document line
    final doc = Map<String, dynamic>.from(produk)..remove('id');
    buffer.writeln(jsonEncode(doc));
  }

  // Send as one request
  final response = await http.post(
    Uri.parse('${es.baseUrl}/_bulk'),
    headers: {
      'Content-Type': 'application/x-ndjson',
      ...es._headers,
    },
    body: buffer.toString(),
  );

  final hasil = jsonDecode(response.body) as Map<String, dynamic>;
  if (hasil['errors'] as bool) {
    final gagal = (hasil['items'] as List)
        .where((item) => (item['index']['status'] as int) >= 400)
        .length;
    print('Bulk index finished with $gagal errors');
  } else {
    print('Bulk index successful: ${produkList.length} documents');
  }
}

Elasticsearch Authentication #

// Elasticsearch basic auth (username/password)
final esBasic = ElasticsearchClient(
  baseUrl: 'https://elasticsearch:9200',
  username: 'elastic',
  password: 'changeme',
);

// API Key (more recommended for production)
final esApiKey = ElasticsearchClient(
  baseUrl: 'https://elasticsearch:9200',
  apiKey: 'base6...==',
);

// Elastic Cloud (Elastic.co hosted)
final esCloud = ElasticsearchClient(
  baseUrl: 'https://my-deployment.es.us-east-1.aws.elastic.cloud:9200',
  apiKey: 'YOUR_API_KEY_HERE',
);

Syncing with the Primary Database #

Elasticsearch is usually used alongside a primary database (PostgreSQL, MySQL) — the primary database as the source of truth, Elasticsearch as the search layer:

import 'package:postgres/postgres.dart';
import 'dart:convert';

// Sync products from PostgreSQL to Elasticsearch
Future<void> sinkronisasiProduk(Pool pgPool, ElasticsearchClient es) async {
  final result = await pgPool.execute(
    'SELECT id, nama, deskripsi, harga, stok, kategori, aktif, dibuat_pada '
    'FROM produk WHERE aktif = true',
  );

  final produkList = result.map((row) {
    final map = row.toColumnMap();
    return {
      'id': map['id'].toString(),
      'nama': map['nama'],
      'deskripsi': map['deskripsi'],
      'harga': map['harga'],
      'stok': map['stok'],
      'kategori': map['kategori'],
      'aktif': map['aktif'],
      'dibuatPada': (map['dibuat_pada'] as DateTime).toIso8601String(),
    };
  }).toList();

  // Bulk index to Elasticsearch
  await bulkIndex(es, produkList);
  print('Synced ${produkList.length} products');
}

Elasticsearch Anti-Patterns #

Using Elasticsearch as the Primary Database #

// ANTI-PATTERN: Elasticsearch isn't suitable as a primary database
// Elasticsearch isn't ACID compliant for all operations
// Newly indexed documents may not be immediately searchable (near real-time)

// ✗ Don't store critical data only in Elasticsearch
await es.put('/transaksi/_doc/T001', transaksiBaru);
// Data may be lost if a node crashes before it's flushed to disk

// CORRECT: Elasticsearch as the search layer, relational/MongoDB as the primary
await pgPool.execute('INSERT INTO transaksi ...');  // primary store
await es.put('/transaksi/_doc/T001', {...});         // secondary search index

Leading Wildcard Queries #

// ANTI-PATTERN: leading wildcard — very slow, performs a full index scan
final buruk = await es.post('/produk/_search', {
  'query': {
    'wildcard': {'nama': '*laptop*'}  // ✗ extremely slow!
  }
});

// CORRECT: use match or multi_match for full-text search
final baik = await es.post('/produk/_search', {
  'query': {
    'match': {'nama': 'laptop'}  // ✓ uses the inverted index
  }
});

Summary #

Elasticsearch uses a REST API — no special driver needed, just the http package. Wrap it in a client class with consistent authentication and error handling.
Explicit mappings are better than auto-detection — define field types like keyword (exact match), text (full-text), date, geo_point before indexing data.
term vs match — term for exact matches on keyword fields (not analyzed), match for full-text search on text fields (tokenized and analyzed).
Bool queries with must (affects the score) and filter (doesn’t affect the score, faster because it’s cached) for complex queries.
fuzziness: 'AUTO' on multi_match provides typo tolerance — ’lapop’ can find ’laptop'.
Aggregations for facets, statistics, and histograms — use size: 0 if you only need aggregations without documents.
The Bulk API for mass indexing — 10-100x faster than individual requests. NDJSON format: one action line + one document line.
Don’t use Elasticsearch as the primary database — use it as a search layer above the primary database (PostgreSQL, MongoDB). Sync data from the primary to Elasticsearch.
Avoid leading wildcards (*keyword) — they perform a full index scan. Use match or multi_match which leverage the inverted index.
keyword fields for sorting, exact-match filtering, and aggregations. text fields for full-text search. Fields needing both: use fields: {keyword: {type: keyword}}.

← Previous: MongoDB Next: Kafka →