Belajar Elasticsearch - Ingest Pipelines & Data Preprocessing

#Pendahuluan

Data mentah di dunia nyata jarang rapi. Log server datang dalam satu string panjang, field tanggal berupa teks bebas, IP tanpa lokasi, dan ada data yang perlu ditambahkan sebelum diindeks. Memaksa aplikasi mengurus ini berarti kode pengirim jadi rumit — padahal solusinya bisa diletakkan di satu tempat: di dalam Elasticsearch itu sendiri.

Ingest pipelines mendefinisikan langkah-langkah preprocessing yang berjalan otomatis saat dokumen masuk. Episode 12 membahas prosesor umum, conditional processing dan pipeline chaining, parsing dengan grok, enrich processor, GeoIP, user agent parser, serta script processor.

#Ingest Node dan Prosesor

Setiap node dengan role ingest mampu menjalankan pipeline. Satu pipeline adalah daftar processors yang dieksekusi berurutan; tiap processor menerima dokumen dan boleh memodifikasinya. Jika satu prosesor gagal, dokumen ditolak — debug dengan _ingest/pipeline/_simulate.

Pipeline dibuat dengan PUT /_ingest/pipeline/nama, lalu dilampirkan saat index dokumen:

{
  "processors": [
    { "set": { "field": "environment", "value": "production" } },
    { "remove": { "field": "sensitive_token" } }
  ]
}

Terapkan lewat PUT /logs/_doc/1?pipeline=normalisasi-log. Dengan pola ini, aplikasi cukup mengirim dokumen mentah dan pipeline-lah yang memastikan konsistensi — semua layanan memakai aturan yang sama.

#Prosesor Umum

{
  "processors": [
    {
      "convert": {
        "field": "http_status",
        "type": "integer",
        "ignore_missing": true
      }
    },
    {
      "rename": {
        "field": "addr",
        "target_field": "source_ip"
      }
    }
  ]
}

Catatan ignore_missing: jika field tidak ada, prosesor tidak error — penting untuk data yang field-nya opsional.

#Conditional Processing dan Pipeline Chaining

Tidak semua dokumen butuh perlakuan yang sama. Prosesor bisa diberi kondisi dengan bahasa Painless sederhana:

{ "set": { "field": "severity_code", "value": 5, "if": "ctx.level == 'error'" } }

Pipeline juga bisa memanggil pipeline lain (chaining) — memecah preprocessing besar menjadi modul yang bisa dipakai ulang:

{
  "processors": [
    { "pipeline": { "name": "normalisasi-field" } },
    { "pipeline": { "name": "enrich-geolokasi" } }
  ]
}

#Parsing Log dengan Grok

Grok mengurai teks semi-terstruktur menjadi field terstruktur memakai pola regex bernama. Contoh paling klasik — log akses Apache/nginx:

{
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": [
          "%{IPORHOST:client_ip} %{USER:ident} %{USER:auth} \\[%{HTTPDATE:timestamp}\\] \"%{WORD:method} %{DATA:uri} %{NOTSPACE:protocol}\" %{NUMBER:status:int} %{NUMBER:bytes:int}"
        ]
      }
    }
  ]
}

Satu baris log 127.0.0.1 - - [03/Aug/2026:10:00:00 +0700] "GET /produk HTTP/1.1" 200 512 langsung berubah menjadi field client_ip, method, uri, status, bytes yang terstruktur. Pola %{IPORHOST}, %{HTTPDATE}, %{WORD} adalah pola bawaan — kalian tidak perlu menulis regex dari nol.

#Enrich, GeoIP, dan User Agent

Enrich processor menggabungkan dokumen masuk dengan data dari index lain. Contoh: log transaksi hanya membawa product_id; kita ingin menambahkan nama dan kategori produk dari index referensi. Enrich butuh enrich policy yang mengatur index sumber, lalu dijalankan:

PUT /_enrich/policy/produk-policy
POST /_enrich/policy/produk-policy/_execute

{
  "processors": [
    {
      "enrich": {
        "policy_name": "produk-policy",
        "field": "product_id",
        "target_field": "produk_detail"
      }
    }
  ]
}

Hasilnya: field produk_detail muncul berisi name dan category — data diperkaya tanpa mengubah aplikasi pengirim.

GeoIP processor memetakan IP menjadi lokasi geografis; user agent processor mengurai string User-Agent menjadi field terstruktur:

{
  "processors": [
    { "geoip": { "field": "source_ip", "target_field": "geo" } },
    { "user_agent": { "field": "ua_string", "target_field": "ua" } }
  ]
}

Dokumen otomatis mendapat geo.country_name, geo.location (tipe geo_point) untuk visualisasi peta, serta ua.name, ua.os.name, ua.device.type untuk analisis perangkat.

#Script Processor

Ada kalanya logika transformasi terlalu spesifik untuk prosesor bawaan. Script processor memakai bahasa Painless untuk transformasi bebas:

{ "script": { "source": "ctx.total_harga = (ctx.harga_satuan ?: 0) * (ctx.jumlah ?: 0); ctx.nama.trim()" } }

Painless adalah bahasa scripting khusus Elasticsearch yang aman dijalankan di server. Untuk logika sederhana gunakan prosesor bawaan; simpan script untuk kasus yang benar-benar butuh fleksibilitas.

#Kesalahan Umum

1. Grok pattern tidak cocok dengan data nyata. Selalu uji dengan _ingest/pipeline/_simulate memakai contoh log asli.
2. Melupakan ignore_missing. Field opsional yang tidak ada membuat prosesor error.
3. Enrich policy tidak di-execute. Policy yang belum dijalankan belum punya index enrich.
4. Pipeline menempel di satu index saja. Lampirkan pipeline ke template (episode 4) agar konsisten.
5. Nested pipeline yang tidak terbatas. Chaining yang terlalu dalam sulit di-debug — batasi tingkatannya.

#Penutup

Di episode 12 kalian menguasai ingest pipelines: prosesor umum (set, remove, rename, convert), conditional processing dengan Painless, pipeline chaining, parsing log dengan grok, enrich processor, GeoIP dan user agent parser, serta script processor untuk logika kustom.

Inti yang harus dibawa pulang:

• Pipeline adalah langkah preprocessing berurutan yang berjalan saat dokumen masuk.
• Grok mengubah string log menjadi field terstruktur dengan pola bawaan.
• Enrich menggabungkan data dari index referensi; jangan lupa execute policy.
• GeoIP dan user agent menambah dimensi lokasi dan perangkat ke data.
• Lampirkan pipeline lewat template agar konsisten di semua index.
• Uji dengan _ingest/pipeline/_simulate sebelum dipakai di produksi.

Seringkali data yang sudah masuk ternyata perlu diubah — mapping keliru, format berubah, atau butuh penggabungan. Di episode 13 kita bahas reindex dan update by query: kapan dan mengapa reindex, reindex dari remote cluster, update by query untuk mass update, penanganan conflict, throttling, serta delete by query. Sampai jumpa!