Elasticsearch

Elasticsearch #

Elasticsearch adalah mesin pencari dan analitik terdistribusi berbasis Apache Lucene — dirancang untuk pencarian full-text yang cepat, analitik log real-time, dan penyimpanan data semi-structured berskala besar. Berbeda dari database lain yang memiliki driver Dart khusus, Elasticsearch diakses melalui REST API standar menggunakan HTTP — ini berarti http package Dart sudah cukup, tanpa dependensi driver tambahan. Artikel ini membahas cara berinteraksi dengan Elasticsearch dari Dart secara efektif, dari manajemen index hingga query DSL yang kompleks.

Konsep Dasar Elasticsearch #

Konsep Elasticsearch Padanan di dunia lain
Index Kumpulan dokumen dengan mapping yang sama Tabel (SQL) / Collection (MongoDB)
Document Satu record JSON dalam index Baris (SQL) / Dokumen (MongoDB)
Field Atribut dalam dokumen Kolom (SQL) / Field (MongoDB)
Shard Potongan horizontal dari index Partition
Replica Salinan shard untuk failover Replica
Mapping Definisi tipe setiap field Schema (SQL)
Query DSL JSON untuk mendefinisikan pencarian SQL WHERE clause
Aggregation Kalkulasi statistik GROUP BY + fungsi agregat SQL

Setup Client #

Karena Elasticsearch menggunakan REST API, tidak ada package driver khusus yang diperlukan. Buat wrapper HTTP client yang reusable:

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,  // contoh: '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';
}

Manajemen Index #

Membuat Index dengan Mapping #

Mapping mendefinisikan tipe setiap field — Elasticsearch bisa auto-detect tipe, tapi eksplisit lebih baik untuk kontrol penuh:

Future<void> buatIndexProduk(ElasticsearchClient es) async {
  final mapping = {
    'settings': {
      'number_of_shards': 1,      // shard untuk produksi kecil-menengah
      'number_of_replicas': 1,    // satu replica untuk HA
      'analysis': {
        'analyzer': {
          'bahasa_indonesia': {
            'type': 'custom',
            'tokenizer': 'standard',
            'filter': ['lowercase', 'asciifolding'],
          }
        }
      }
    },
    'mappings': {
      'properties': {
        'nama': {
          'type': 'text',
          'analyzer': 'bahasa_indonesia',
          'fields': {
            'keyword': {'type': 'keyword'},  // untuk sorting dan exact match
          }
        },
        'deskripsi': {'type': 'text', 'analyzer': 'bahasa_indonesia'},
        'harga': {'type': 'double'},
        'stok': {'type': 'integer'},
        'kategori': {'type': 'keyword'},    // keyword = exact match, tidak di-tokenize
        'tag': {'type': 'keyword'},
        'aktif': {'type': 'boolean'},
        'dibuatPada': {'type': 'date'},
        'lokasi': {'type': 'geo_point'},    // untuk pencarian berdasarkan lokasi
        'spesifikasi': {'type': 'object'},  // nested JSON object
      }
    }
  };

  try {
    await es.put('/produk', mapping);
    print('Index produk berhasil dibuat');
  } on ElasticsearchException catch (e) {
    if (e.statusCode == 400 && e.tipe == 'resource_already_exists_exception') {
      print('Index sudah ada, lewati pembuatan');
    } else {
      rethrow;
    }
  }
}

// Cek apakah index ada
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;
  }
}

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

CRUD Dokumen #

Index (Insert/Upsert) # 

Future<void> contohIndex(ElasticsearchClient es) async {
  // Index dengan ID yang ditentukan — upsert (create atau replace)
  final hasilDenganId = await es.put('/produk/_doc/P001', {
    'nama': 'Laptop Gaming ASUS ROG',
    'deskripsi': 'Laptop gaming performa tinggi dengan 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('Hasil index: ${hasilDenganId['result']}'); // created atau updated

  // Index tanpa ID — Elasticsearch generate ID otomatis
  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('ID yang dibuat: ${hasilTanpaId['_id']}');

  // _create — hanya buat jika belum ada (gagal jika ID sudah ada)
  await es.put('/produk/_create/P999', {
    'nama': 'Produk Baru',
    'harga': 100_000.0,
    'aktif': true,
  });
}

Get Dokumen # 

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 — ambil beberapa dokumen sekaligus
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();
}

Update Dokumen # 

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

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

  // Upsert — update jika ada, insert jika tidak ada
  await es.post('/produk/_update/P002', {
    'doc': {'stok': 0, 'aktif': false},
    'upsert': {  // digunakan jika dokumen belum ada
      'nama': 'Produk Default',
      'stok': 0,
      'aktif': false,
    }
  });
}

Delete Dokumen # 

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

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

Query DSL — Pencarian yang Powerful #

Query Dasar # 

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',  // toleransi typo: 'lapop' → 'laptop'
      }
    },
    'size': 20,
    'from': 0,
    'highlight': {  // tampilkan bagian yang cocok
      'fields': {
        'nama': {},
        'deskripsi': {'fragment_size': 150},
      }
    },
    '_source': ['nama', 'harga', 'kategori', 'stok'],  // pilih field yang dikembalikan
  };

  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 Query — Kombinasi Kondisi # 

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 — mempengaruhi relevance score
  if (keyword != null && keyword.isNotEmpty) {
    mustClauses.add({
      'multi_match': {
        'query': keyword,
        'fields': ['nama^3', 'deskripsi', 'tag'],
        'fuzziness': 'AUTO',
      }
    });
  }

  // Filter — tidak mempengaruhi score, lebih cepat karena di-cache
  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'}]  // relevan dulu, lalu harga murah
        : [{'dibuatPada': 'desc'}],      // tanpa keyword: terbaru dulu
    'size': perHalaman,
    'from': (halaman - 1) * perHalaman,
    'track_total_hits': true,
  });

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

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

Aggregation — Statistik dan Facet # 

Future<Map<String, dynamic>> statistikProduk(ElasticsearchClient es) async {
  final hasil = await es.post('/produk/_search', {
    'size': 0,  // tidak perlu dokumen, hanya aggregasi
    'query': {'term': {'aktif': true}},
    'aggs': {
      // Jumlah per kategori (facet)
      'perKategori': {
        'terms': {
          'field': 'kategori',
          'size': 20,
          'order': {'_count': 'desc'},
        }
      },
      // Statistik harga
      'statsHarga': {
        'stats': {'field': 'harga'}
      },
      // Harga rata-rata per kategori
      'hargaRataPerKategori': {
        'terms': {'field': 'kategori'},
        'aggs': {
          'rataHarga': {'avg': {'field': 'harga'}},
          'totalStok': {'sum': {'field': 'stok'}},
        }
      },
      // Histogram harga — distribusi produk per rentang harga
      '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>;

  // Proses hasil aggregasi
  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 #

Untuk mengindex banyak dokumen sekaligus, gunakan Bulk API — jauh lebih efisien dari request individual:

Future<void> bulkIndex(
    ElasticsearchClient es, List<Map<String, dynamic>> produkList) async {
  // Bulk API menggunakan format NDJSON (newline-delimited JSON)
  // Setiap operasi: satu baris action + satu baris dokumen
  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));
  }

  // Kirim sebagai satu 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 selesai dengan $gagal error');
  } else {
    print('Bulk index berhasil: ${produkList.length} dokumen');
  }
}

Autentikasi Elasticsearch # 

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

// API Key (lebih direkomendasikan untuk produksi)
final esApiKey = ElasticsearchClient(
  baseUrl: 'https://elasticsearch:9200',
  apiKey: 'base64EncodedApiKey==',
);

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

Sinkronisasi dengan Database Utama #

Elasticsearch biasanya digunakan bersama database utama (PostgreSQL, MySQL) — database utama sebagai source of truth, Elasticsearch sebagai search layer:

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

// Sinkronisasi produk dari PostgreSQL ke 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 ke Elasticsearch
  await bulkIndex(es, produkList);
  print('Sinkronisasi ${produkList.length} produk selesai');
}

Anti-Pattern Elasticsearch #

Menggunakan Elasticsearch sebagai Database Utama # 

// ANTI-PATTERN: Elasticsearch tidak cocok sebagai primary database
// Elasticsearch tidak ACID compliant untuk semua operasi
// Dokumen yang baru diindex mungkin tidak langsung bisa di-search (near real-time)

// ✗ Jangan simpan data kritis hanya di Elasticsearch
await es.put('/transaksi/_doc/T001', transaksiBaru);
// Data mungkin hilang jika node crash sebelum di-flush ke disk

// BENAR: Elasticsearch sebagai layer pencarian, database relasional/MongoDB sebagai primary
await pgPool.execute('INSERT INTO transaksi ...');  // primary store
await es.put('/transaksi/_doc/T001', {...});         // secondary search index

Query Wildcard di Awal String # 

// ANTI-PATTERN: wildcard di awal — sangat lambat, melakukan full index scan
final buruk = await es.post('/produk/_search', {
  'query': {
    'wildcard': {'nama': '*laptop*'}  // ✗ extremely slow!
  }
});

// BENAR: gunakan match atau multi_match untuk full-text search
final baik = await es.post('/produk/_search', {
  'query': {
    'match': {'nama': 'laptop'}  // ✓ menggunakan inverted index
  }
});

Ringkasan #

  • Elasticsearch menggunakan REST API — tidak perlu driver khusus, cukup http package. Bungkus dalam client class dengan autentikasi dan error handling yang konsisten.
  • Mapping eksplisit lebih baik dari auto-detect — definisikan tipe field seperti keyword (exact match), text (full-text), date, geo_point sebelum index data.
  • term vs match — term untuk exact match pada field keyword (tidak di-analisis), match untuk full-text search pada field text (di-tokenize dan di-analisis).
  • Bool query dengan must (mempengaruhi score) dan filter (tidak mempengaruhi score, lebih cepat karena di-cache) untuk query yang kompleks.
  • fuzziness: 'AUTO' pada multi_match memberikan toleransi typo — ’lapop’ bisa menemukan ’laptop'.
  • Aggregation untuk facet, statistik, dan histogram — gunakan size: 0 jika hanya butuh aggregasi tanpa dokumen.
  • Bulk API untuk indexing massal — 10-100x lebih cepat dari request individual. Format NDJSON: satu baris action + satu baris dokumen.
  • Jangan gunakan Elasticsearch sebagai primary database — gunakan sebagai search layer di atas database utama (PostgreSQL, MongoDB). Sinkronisasi data dari primary ke Elasticsearch.
  • Hindari wildcard di awal string (*keyword) — melakukan full index scan. Gunakan match atau multi_match yang memanfaatkan inverted index.
  • Field keyword untuk sorting, filtering exact match, dan aggregation. Field text untuk full-text search. Field yang butuh keduanya: gunakan fields: {keyword: {type: keyword}}.

← Sebelumnya: MongoDB   Berikutnya: Kafka →

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